#!/usr/bin/env python3 # -*- coding: utf-8 -*- from __future__ import annotations import unittest import zlib from core.ksse_sted import ( DFP_SUFFIX_LEN, KSSE_DFP_TABLE_RAW, STED_CACHE_FILE_NAME, STED_SHARED_PREF_KEY, KsseDfpStringTable, KsseSentinelMissing, build_sted_cache_json, build_sted_persistence_artifacts, build_sted_result_json, dfp_suffix_from_value, engine_sted_product_marker, hidden_cache_path, ksse_dfp_suffix_crc32, ksse_dfp_suffix_crc_material, ksse_md5_hex16, ksse_suffix_crc32, ksse_suffix_crc_material, normalize_product_marker, recover_first_dfp_value_from_candidate_paths, recover_dfp_suffix_from_sted_sentinels, recover_dfp_value_from_sted_sentinels, recover_suffix_from_sentinels, recover_sted_result_json_from_sentinels, recover_sted_values_from_sentinels, select_sted_product_value, sentinel_char_path, sentinel_half, sentinel_half_path, sentinel_paths_for_suffix, sted_candidate_path_families, sted_external_candidate_paths, sted_hidden_md5_candidate_paths, sted_json_insertion_plan, sted_product_salt_candidate_paths, sted_crc_marker_path, sted_writer_marker_suffix, sted_writer_paths, sted_writer_sentinel_paths, ) class KsseStedModelTests(unittest.TestCase): def test_table_layout_matches_static_decode(self) -> None: table = KsseDfpStringTable.parse() self.assertEqual(table.alphabet, "0123456789ABCDEF") self.assertEqual(table.android_hidden_dir_suffix, "/.Android_") self.assertEqual(table.default_product, "KUAISHOU") self.assertEqual(table.static_salt, "a21c40ada1eb475b") self.assertEqual(table.dfp_prefix, "DFP") self.assertEqual(table.sdcard_android_dir, "/sdcard/Android") self.assertEqual(table.active_product, "NEBULA") def test_sentinel_path_naming(self) -> None: base = "/sdcard/Documents" self.assertEqual(sentinel_half(1, "0"), "0") self.assertEqual(sentinel_half(1, "7"), "0") self.assertEqual(sentinel_half(1, "8"), "1") self.assertEqual(sentinel_half(1, "F"), "1") self.assertEqual(sentinel_half_path(base, 12, "0"), "/sdcard/Documents/.12@0") self.assertEqual(sentinel_char_path(base, 12, "A"), "/sdcard/Documents/.12_A") def test_writer_paths_include_half_and_exact_markers(self) -> None: paths = sentinel_paths_for_suffix("/tmp/base/", "07AF") self.assertEqual( paths, [ "/tmp/base/.1@0", "/tmp/base/.1_0", "/tmp/base/.2@0", "/tmp/base/.2_7", "/tmp/base/.3@1", "/tmp/base/.3_A", "/tmp/base/.4@1", "/tmp/base/.4_F", ], ) def test_recover_suffix_uses_existing_exact_marker(self) -> None: suffix = "0123456789ABCDEF" * 4 suffix = suffix[:DFP_SUFFIX_LEN] existing = set(sentinel_paths_for_suffix("/base", suffix)) recovered = recover_suffix_from_sentinels("/base", existing.__contains__) self.assertEqual(recovered, suffix) def test_recover_suffix_falls_back_to_full_alphabet_without_half_marker(self) -> None: suffix = "F" * DFP_SUFFIX_LEN existing = { path for path in sentinel_paths_for_suffix("/base", suffix) if "@" not in path } recovered = recover_suffix_from_sentinels("/base", existing.__contains__) self.assertEqual(recovered, suffix) def test_recover_suffix_missing_marker_raises(self) -> None: with self.assertRaises(KsseSentinelMissing): recover_suffix_from_sentinels("/base", lambda _path: False) def test_crc_material_matches_static_order(self) -> None: suffix = "ABCDEF" + "0" * (DFP_SUFFIX_LEN - 6) model = "OnePlus(PJZ110)" material = ksse_suffix_crc_material(suffix, model=model) self.assertEqual(material.decode(), KSSE_DFP_TABLE_RAW + suffix[::-1] + model) self.assertEqual(ksse_suffix_crc32(suffix, model=model), zlib.crc32(material) & 0xFFFFFFFF) def test_product_marker_and_json_selection(self) -> None: data = build_sted_result_json( { "NEBULA": "DFP" + "A" * DFP_SUFFIX_LEN, "KUAISHOU": "DFP" + "B" * DFP_SUFFIX_LEN, } ) self.assertEqual(normalize_product_marker("0NEBULA"), "NEBULA") self.assertEqual(normalize_product_marker("1KWE_N"), "KWE_N") self.assertEqual(select_sted_product_value(data, "0NEBULA"), "DFP" + "A" * DFP_SUFFIX_LEN) self.assertEqual(select_sted_product_value(data, "0MISSING"), "DFP" + "B" * DFP_SUFFIX_LEN) def test_md5_hex16_uses_lowercase_first_eight_digest_bytes(self) -> None: self.assertEqual(ksse_md5_hex16("NEBULA"), "f0fcd2bde58f4ae8") self.assertEqual(ksse_md5_hex16(b"KUAISHOU"), "b00f2012d71ddb04") def test_engine_sted_product_marker_matches_java_z_flag(self) -> None: self.assertEqual(engine_sted_product_marker("NEBULA", writable_external_storage=False), "0NEBULA") self.assertEqual(engine_sted_product_marker("NEBULA", writable_external_storage=True), "1NEBULA") self.assertEqual(engine_sted_product_marker("0KUAISHOU"), "0KUAISHOU") def test_build_sted_cache_json_matches_rq0_d_e_shape(self) -> None: self.assertEqual( build_sted_cache_json("DFP" + "A" * DFP_SUFFIX_LEN, "cache_m", 1780000000123), '{"c_time":1780000000123,"cache_e":"DFP' + "A" * DFP_SUFFIX_LEN + '","cache_m":"cache_m"}', ) def test_hidden_cache_path_joins_native_hidden_suffix(self) -> None: self.assertEqual( hidden_cache_path("/sdcard/Documents/", "abc123"), "/sdcard/Documents/.Android_abc123", ) def test_external_candidates_use_product_dependent_suffix(self) -> None: self.assertEqual( sted_external_candidate_paths("0KUAISHOU"), [ "/sdcard/Documents/.Android_a21c40ada1eb475b", "/sdcard/Android/.Android_a21c40ada1eb475b", ], ) self.assertEqual( sted_external_candidate_paths("0NEBULA"), [ "/sdcard/Documents/.Android_f0fcd2bde58f4ae8", "/sdcard/Android/.Android_f0fcd2bde58f4ae8", ], ) def test_product_salt_candidates_are_documents_then_android(self) -> None: self.assertEqual( sted_product_salt_candidate_paths(), [ "/sdcard/Documents/.Android_a21c40ada1eb475b", "/sdcard/Android/.Android_a21c40ada1eb475b", ], ) def test_hidden_md5_candidates_hash_active_table_product(self) -> None: self.assertEqual( sted_hidden_md5_candidate_paths(), [ "/sdcard/Documents/.Android_f0fcd2bde58f4ae8", "/sdcard/Android/.Android_f0fcd2bde58f4ae8", ], ) self.assertEqual( sted_hidden_md5_candidate_paths(active_product="KWE_N"), [ "/sdcard/Documents/.Android_cbaa84d8122aca37", "/sdcard/Android/.Android_cbaa84d8122aca37", ], ) def test_candidate_path_families_group_recovered_order(self) -> None: families = sted_candidate_path_families("0NEBULA") self.assertEqual(list(families), ["external", "product_salt", "hidden_md5"]) self.assertEqual( families["external"], [ "/sdcard/Documents/.Android_f0fcd2bde58f4ae8", "/sdcard/Android/.Android_f0fcd2bde58f4ae8", ], ) self.assertEqual( families["product_salt"], [ "/sdcard/Documents/.Android_a21c40ada1eb475b", "/sdcard/Android/.Android_a21c40ada1eb475b", ], ) def test_json_insertion_plan_for_active_product(self) -> None: plan = sted_json_insertion_plan("0NEBULA") self.assertEqual([item.output_key for item in plan], ["NEBULA", "KUAISHOU"]) self.assertEqual([item.stage for item in plan], ["external_current", "product_salt_default"]) self.assertIn("bVar6", plan[1].guard) def test_json_insertion_plan_for_default_product(self) -> None: plan = sted_json_insertion_plan("0KUAISHOU") self.assertEqual([item.output_key for item in plan], ["KUAISHOU", "NEBULA"]) self.assertEqual([item.stage for item in plan], ["external_current", "hidden_md5_active"]) self.assertIn("bVar1", plan[1].guard) def test_json_insertion_plan_for_other_product(self) -> None: plan = sted_json_insertion_plan("0KWE_N") self.assertEqual([item.output_key for item in plan], ["KWE_N", "KUAISHOU", "NEBULA"]) self.assertEqual( [item.stage for item in plan], ["external_current", "product_salt_default", "hidden_md5_active"], ) self.assertIn("cbaa84d8122aca37", plan[0].candidate_paths[0]) self.assertIn("!bVar1 && !bVar6", plan[2].guard) def test_writer_strips_dfp_prefix_and_reverses_suffix(self) -> None: suffix = ("0123456789ABCDEF" * 4)[:DFP_SUFFIX_LEN] value = "DFP" + suffix self.assertEqual(dfp_suffix_from_value(value), suffix) self.assertEqual(dfp_suffix_from_value(suffix), suffix) self.assertEqual(sted_writer_marker_suffix(value), suffix[::-1]) def test_writer_paths_include_reversed_sentinels_and_crc_marker(self) -> None: suffix = ("0123456789ABCDEF" * 4)[:DFP_SUFFIX_LEN] value = "DFP" + suffix base = "/sdcard/Documents/.Android_f0fcd2bde58f4ae8" model = "OnePlus(PJZ110)" self.assertEqual( sted_writer_sentinel_paths(base, value), sentinel_paths_for_suffix(base, suffix[::-1]), ) self.assertEqual( ksse_dfp_suffix_crc_material(suffix, model=model).decode(), KSSE_DFP_TABLE_RAW + suffix + model, ) self.assertEqual( sted_crc_marker_path(base, value, model=model), f"{base}/.{ksse_dfp_suffix_crc32(suffix, model=model)}", ) self.assertEqual( sted_writer_paths(base, value, model=model)[-1], sted_crc_marker_path(base, value, model=model), ) def test_recover_dfp_value_from_writer_sentinels(self) -> None: suffix = ("FEDCBA9876543210" * 4)[:DFP_SUFFIX_LEN] value = "DFP" + suffix base = "/base" model = "ModelX" existing = set(sted_writer_paths(base, value, model=model)) self.assertEqual( recover_suffix_from_sentinels(base, existing.__contains__), suffix[::-1], ) self.assertEqual( recover_dfp_suffix_from_sted_sentinels( base, existing.__contains__, model=model, require_crc=True, ), suffix, ) self.assertEqual( recover_dfp_value_from_sted_sentinels( base, existing.__contains__, model=model, require_crc=True, ), value, ) def test_crc_guard_can_be_required(self) -> None: suffix = "A" * DFP_SUFFIX_LEN value = "DFP" + suffix existing = set(sted_writer_sentinel_paths("/base", value)) with self.assertRaises(KsseSentinelMissing): recover_dfp_suffix_from_sted_sentinels( "/base", existing.__contains__, require_crc=True, ) def test_recover_first_candidate_uses_primary_then_fallback(self) -> None: value = "DFP" + "B" * DFP_SUFFIX_LEN primary = "/primary" fallback = "/fallback" existing = set(sted_writer_paths(fallback, value)) self.assertEqual( recover_first_dfp_value_from_candidate_paths( [primary, fallback], existing.__contains__, ), value, ) def test_recover_sted_values_for_active_product(self) -> None: nebula = "DFP" + "1" * DFP_SUFFIX_LEN kuaishou = "DFP" + "2" * DFP_SUFFIX_LEN plan = sted_json_insertion_plan("0NEBULA") existing = set(sted_writer_paths(plan[0].candidate_paths[0], nebula)) existing.update(sted_writer_paths(plan[1].candidate_paths[0], kuaishou)) values = recover_sted_values_from_sentinels("0NEBULA", existing.__contains__) self.assertEqual(list(values), ["NEBULA", "KUAISHOU"]) self.assertEqual(values["NEBULA"], nebula) self.assertEqual(values["KUAISHOU"], kuaishou) self.assertEqual( recover_sted_result_json_from_sentinels("0NEBULA", existing.__contains__), build_sted_result_json(values), ) def test_recover_sted_values_for_default_product_uses_fallback_paths(self) -> None: kuaishou = "DFP" + "3" * DFP_SUFFIX_LEN nebula = "DFP" + "4" * DFP_SUFFIX_LEN plan = sted_json_insertion_plan("0KUAISHOU") existing = set(sted_writer_paths(plan[0].candidate_paths[1], kuaishou)) existing.update(sted_writer_paths(plan[1].candidate_paths[1], nebula)) values = recover_sted_values_from_sentinels("0KUAISHOU", existing.__contains__) self.assertEqual(list(values), ["KUAISHOU", "NEBULA"]) self.assertEqual(values["KUAISHOU"], kuaishou) self.assertEqual(values["NEBULA"], nebula) def test_recover_sted_values_for_other_product_preserves_native_order(self) -> None: kwe_n = "DFP" + "5" * DFP_SUFFIX_LEN kuaishou = "DFP" + "6" * DFP_SUFFIX_LEN nebula = "DFP" + "7" * DFP_SUFFIX_LEN plan = sted_json_insertion_plan("0KWE_N") existing = set(sted_writer_paths(plan[0].candidate_paths[0], kwe_n)) existing.update(sted_writer_paths(plan[1].candidate_paths[0], kuaishou)) existing.update(sted_writer_paths(plan[2].candidate_paths[0], nebula)) values = recover_sted_values_from_sentinels("0KWE_N", existing.__contains__) self.assertEqual(list(values), ["KWE_N", "KUAISHOU", "NEBULA"]) self.assertEqual(values["KWE_N"], kwe_n) self.assertEqual(values["KUAISHOU"], kuaishou) self.assertEqual(values["NEBULA"], nebula) def test_build_sted_persistence_artifacts_models_rq0_d_e_chain(self) -> None: value = "DFP" + "8" * DFP_SUFFIX_LEN artifacts = build_sted_persistence_artifacts( value, "cache_m_value", c_time_ms=1780000000123, product="NEBULA", writable_external_storage=False, model="OnePlus(PJZ110)", ) self.assertEqual(artifacts.product_marker, "0NEBULA") self.assertEqual( artifacts.cache_json, '{"c_time":1780000000123,"cache_e":"' + value + '","cache_m":"cache_m_value"}', ) self.assertEqual(artifacts.in_memory_cache["cache_e"], value) self.assertEqual(artifacts.in_memory_cache["cache_m"], "cache_m_value") self.assertEqual(artifacts.in_memory_cache["c_time"], "1780000000123") self.assertEqual(artifacts.shared_preferences[STED_SHARED_PREF_KEY], artifacts.cache_json) self.assertEqual(artifacts.app_private_files[STED_CACHE_FILE_NAME], artifacts.cache_json) self.assertEqual( artifacts.native_base_path, "/sdcard/Documents/.Android_f0fcd2bde58f4ae8", ) self.assertIn(sted_crc_marker_path(artifacts.native_base_path, value, model="OnePlus(PJZ110)"), artifacts.native_sentinel_paths) self.assertEqual( artifacts.native_readback_json, build_sted_result_json({"NEBULA": value}), ) if __name__ == "__main__": unittest.main()