ksjsb/tests/test_ksse_sted.py
2026-07-30 20:25:56 +08:00

413 lines
16 KiB
Python

#!/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()