"""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 _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, 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(" 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("= 16: cfs = i - 16 length = struct.unpack_from(" 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("> 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("> 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("> 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("> 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("> 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(" 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("= 4 and _is_pacibsp(struct.unpack_from("= 8: ins_b = struct.unpack_from(" len(blob): break if struct.unpack_from(" 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(" 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(" 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(" 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)